MQ!Rabbit-client ChannelN
引
MQ!Rabbit-client command中有一个小尾巴没有处理,看下面的代码
public void handleFrame(Frame frame) throws IOException {
AMQCommand command = _command;
if (command.handleFrame(frame)) { // a complete command has rolled off the assembly line
_command = new AMQCommand(); // prepare for the next one
handleCompleteInboundCommand(command);
}
}
这段代码是rabbitmq处理帧的操作，当时分析了command.handleFrame(frame)这个方法实际上就是调用CommandAssembler.handleFrame()。这里我们来下面的方法handleCompleteInboundCommand(command);
// handle a command which has been assembled
// 处理已组装好的命令
public void handleCompleteInboundCommand(AMQCommand command) throws IOException {
// First, offer the command to the asynchronous-command
// handling mechanism, which gets to act as a filter on the
// incoming command stream. If processAsync() returns true,
// the command has been dealt with by the filter and so should
// not be processed further. It will return true for
// asynchronous commands (deliveries/returns/other events),
// and false for commands that should be passed on to some
// waiting RPC continuation.
// 过滤掉已经处理的命令
if (!processAsync(command)) {
// The filter decided not to handle/consume the command,
// so it must be a response to an earlier RPC.
if (_checkRpcResponseType) {
synchronized (_channelMutex) {
// check if this reply command is intended for the current waiting request before calling nextOutstandingRpc()
if (_activeRpc != null && !_activeRpc.canHandleReply(command)) {
// this reply command is not intended for the current waiting request
// most likely a previous request timed out and this command is the reply for that.
// Throw this reply command away so we don't stop the current request from waiting for its reply
return;
}
}
}
final RpcWrapper nextOutstandingRpc = nextOutstandingRpc();
// the outstanding RPC can be null when calling Channel#asyncRpc
if(nextOutstandingRpc != null) {
nextOutstandingRpc.complete(command);
markRpcFinished();
}
}
}
敲黑板了！！！🐷
processAsync这个方法的实现是在ChnnelN中(终于绕到ChannelN上了 😄)。
另外这个方法也是AMQChannel中唯一的抽象方法
正
ChannelN是整个RabbitMQ客户端最核心的一个类了✨
ChannelN的成员变量
private final Map<String, Consumer> _consumers = Collections.synchronizedMap(new HashMap<String, Consumer>());
private volatile Consumer defaultConsumer = null;
private final ConsumerDispatcher dispatcher;
private final Collection<ReturnListener> returnListeners = new CopyOnWriteArrayList<ReturnListener>();
private final Collection<FlowListener> flowListeners = new CopyOnWriteArrayList<FlowListener>();
private volatile CountDownLatch finishedShutdownFlag = null;
private final Collection<ConfirmListener> confirmListeners = new CopyOnWriteArrayList<ConfirmListener>();
private long nextPublishSeqNo = 0L;
private final SortedSet<Long> unconfirmedSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
private volatile boolean onlyAcksReceived = true;
processAsync方法
原文注释都被我干掉了🙃
@Override public boolean processAsync(Command command) throws IOException
{
Method method = command.getMethod();
// 方法帧 Channel.Close，异步关闭
if (method instanceof Channel.Close) {
asyncShutdown(command);
return true;
}
// 链接打开中
if (isOpen()) {
if (method instanceof Basic.Deliver) {
processDelivery(command, (Basic.Deliver) method);
return true;
} else if (method instanceof Basic.Return) {
callReturnListeners(command, (Basic.Return) method);
return true;
} else if (method instanceof Channel.Flow) {
Channel.Flow channelFlow = (Channel.Flow) method;
synchronized (_channelMutex) {
_blockContent = !channelFlow.getActive();
transmit(new Channel.FlowOk(!_blockContent));
_channelMutex.notifyAll();
}
return true;
} else if (method instanceof Basic.Ack) {
Basic.Ack ack = (Basic.Ack) method;
callConfirmListeners(command, ack);
handleAckNack(ack.getDeliveryTag(), ack.getMultiple(), false);
return true;
} else if (method instanceof Basic.Nack) {
Basic.Nack nack = (Basic.Nack) method;
callConfirmListeners(command, nack);
handleAckNack(nack.getDeliveryTag(), nack.getMultiple(), true);
return true;
} else if (method instanceof Basic.RecoverOk) {
for (Map.Entry<String, Consumer> entry : Utility.copy(_consumers).entrySet()) {
this.dispatcher.handleRecoverOk(entry.getValue(), entry.getKey());
}
return false;
} else if (method instanceof Basic.Cancel) {
Basic.Cancel m = (Basic.Cancel)method;
String consumerTag = m.getConsumerTag();
Consumer callback = _consumers.remove(consumerTag);
if (callback == null) {
callback = defaultConsumer;
}
if (callback != null) {
try {
this.dispatcher.handleCancel(callback, consumerTag);
} catch (Throwable ex) {
getConnection().getExceptionHandler().
handleConsumerException(this, ex, callback,consumerTag,"handleCancel");
}
}
return true;
} else {
return false;
}
} else {
// Channel.CloseOk方法
if (method instanceof Channel.CloseOk) {
return false;
} else {
return true;
}
}
}
这个方法主要用来针对接受到broker的AMQCommand进行进一步的处理，至于怎么接受Socket，怎么封装成帧，怎么确定一个AMQComand已经封装完毕，都已在调用此方法前完成。此方法可以处理：Channel.Close, Basic.Deliver, Basic.Return, Channel.Flow, Basic.Ack, Basic.Nack,Basic.RecoverOk, Basic.Cancel, Channel.CloseOk等。
那来吧开始打怪了🤢。
// Basic.Deliver
processDelivery(command, (Basic.Deliver) method);
protected void processDelivery(Command command, Basic.Deliver method) {
Basic.Deliver m = method;
Consumer callback = _consumers.get(m.getConsumerTag());
if (callback == null) {
if (defaultConsumer == null) {
// No handler set. We should blow up as this message
// needs acking, just dropping it is not enough. See bug
// 22587 for discussion.
throw new IllegalStateException("Unsolicited delivery -" +
" see Channel.setDefaultConsumer to handle this" +
" case.");
} else {
callback = defaultConsumer;
}
}
Envelope envelope = new Envelope(m.getDeliveryTag(),
m.getRedelivered(),
m.getExchange(),
m.getRoutingKey());
try {
// call metricsCollector before the dispatching (which is async anyway)
// this way, the message is inside the stats before it is handled
// in case a manual ack in the callback, the stats will be able to record the ack
metricsCollector.consumedMessage(this, m.getDeliveryTag(), m.getConsumerTag());
// ConsumerWorkService实例
this.dispatcher.handleDelivery(callback,
m.getConsumerTag(),
envelope,
(BasicProperties) command.getContentHeader(),
command.getContentBody());
} catch (Throwable ex) {
getConnection().getExceptionHandler().handleConsumerException(this,
ex,
callback,
m.getConsumerTag(),
"handleDelivery");
}
}
在学习ConsumerWorkService里面留了一个尾巴就是ConsumerDispatcher。这里会进行补充吧。分析下上面代码里面的this.dispatcher.handleDelivery
public void handleDelivery(final Consumer delegate,
final String consumerTag,
final Envelope envelope,
final AMQP.BasicProperties properties,
final byte[] body) throws IOException {
executeUnlessShuttingDown(
new Runnable() {
@Override
public void run() {
try {
delegate.handleDelivery(consumerTag,
envelope,
properties,
body);
} catch (Throwable ex) {
connection.getExceptionHandler().handleConsumerException(
channel,
ex,
delegate,
consumerTag,
"handleDelivery");
}
}
});
}
不要看里面Runnable的实现看外面套的那层方法：
private void executeUnlessShuttingDown(Runnable r) {
if (!this.shuttingDown) execute(r);
}
private void execute(Runnable r) {
checkShutdown();
this.workService.addWork(this.channel, r);
}
这里就可以看到最后其实还是将需要处理的任务封装为一个Runnable然后调用了workService.addWork。
所以说其实ConsumerDispatcher其实就是把任务分发给workPool处理。
什么？不知道workService.addWork是干什么的！ 嘿嘿，去帮我刷下浏览量吧😜 MQ!Rabbit-client ConsumerWorkService
★☆★☆★☆★☆★☆★☆ ⊙◎○●”·． 可爱的分割线 ．·“●○◎⊙ ★☆★☆★☆★☆★☆★☆
上面已经跑偏了，这里主要是要学习ChannelN的。来继续看下其他方法吧
Basic.Qos & basicQos
@Override
public void basicQos(int prefetchSize, int prefetchCount, boolean global)
throws IOException{
exnWrappingRpc(new Basic.Qos(prefetchSize, prefetchCount, global));
}
// AMQImpl
public Qos(int prefetchSize, int prefetchCount, boolean global) {
this.prefetchSize = prefetchSize;
this.prefetchCount = prefetchCount;
this.global = global;
}
好吧，这个源码我是看不出来有什么了，只好找网上的说法了。
消费者在开启ACK的情况下，对接受到的消息可以根据业务的需要异步对消息进行确认。
然而在实际使用过程中，由于消费者自身处理能力有限，从RabbitMQ获取一定数量的消息后，希望rabbitmq不再将队列中的消息推送过来，当对消息处理完后（即对消息进行了ack,并且有能力处理更多的消息）再接受来自队列的消息。在这种场景下，我们可以设置Basic.Qos中的prefetch_count来达到这个效果。
Basic.Consume & basicConsume
发送Basic.Consume帧，然后等待Basic.ConsumeOk帧。待收到broker端的Basic.ConsumeOk帧之后，触发BlockingRpcContinuation中的transformReply()方法
当发送Basic.Consume帧之后，由broker返回的是Basic.ConsumeOk帧+Basic.Deliver帧，Basic.ConsumerOk帧由下面方法处理，Basic.Deliver帧由processAsync处理。
@Override
public String basicConsume(String queue, final boolean autoAck, String consumerTag,
boolean noLocal, boolean exclusive, Map<String, Object> arguments,
final Consumer callback)
throws IOException {
final Method m = new Basic.Consume.Builder()
.queue(queue)
.consumerTag(consumerTag)
.noLocal(noLocal)
.noAck(autoAck)
.exclusive(exclusive)
.arguments(arguments)
.build();
BlockingRpcContinuation<String> k = new BlockingRpcContinuation<String>(m) {
@Override
public String transformReply(AMQCommand replyCommand) {
String actualConsumerTag = ((Basic.ConsumeOk) replyCommand.getMethod()).getConsumerTag();
_consumers.put(actualConsumerTag, callback);
// need to register consumer in stats before it actually starts consuming
metricsCollector.basicConsume(ChannelN.this, actualConsumerTag, autoAck);
dispatcher.handleConsumeOk(callback, actualConsumerTag);
return actualConsumerTag;
}
};
rpc(m, k);
try {
if(_rpcTimeout == NO_RPC_TIMEOUT) {
return k.getReply();
} else {
try {
return k.getReply(_rpcTimeout);
} catch (TimeoutException e) {
throw wrapTimeoutException(m, e);
}
}
} catch(ShutdownSignalException ex) {
throw wrap(ex);
}
}
Basic.Get & basicGet
基本上就是客户端发送Basic.Get至Broker，Broker返回Basic.GetOK并携带数据。注意方法最后返回GetResponse对象，这个对象就是包装了一下数据。
public GetResponse basicGet(String queue, boolean autoAck)
throws IOException
{
AMQCommand replyCommand = exnWrappingRpc(new Basic.Get.Builder()
.queue(queue)
.noAck(autoAck)
.build());
Method method = replyCommand.getMethod();
if (method instanceof Basic.GetOk) {
Basic.GetOk getOk = (Basic.GetOk)method;
Envelope envelope = new Envelope(getOk.getDeliveryTag(),
getOk.getRedelivered(),
getOk.getExchange(),
getOk.getRoutingKey());
BasicProperties props = (BasicProperties)replyCommand.getContentHeader();
byte[] body = replyCommand.getContentBody();
int messageCount = getOk.getMessageCount();
return new GetResponse(envelope, props, body, messageCount);
} else if (method instanceof Basic.GetEmpty) {
return null;
} else {
throw new UnexpectedMethodError(method);
}
}
事务
/** Public API - {@inheritDoc} */
public Tx.SelectOk txSelect()
throws IOException
{
return (Tx.SelectOk) exnWrappingRpc(new Tx.Select()).getMethod();
}
/** Public API - {@inheritDoc} */
public Tx.CommitOk txCommit()
throws IOException
{
return (Tx.CommitOk) exnWrappingRpc(new Tx.Commit()).getMethod();
}
/** Public API - {@inheritDoc} */
public Tx.RollbackOk txRollback()
throws IOException
{
return (Tx.RollbackOk) exnWrappingRpc(new Tx.Rollback()).getMethod();
}
上面几个方法参考博客：
