RabbitMQ-客户端源码之ChannelManager
创始人
2024-05-25 09:58:43
0

关于ChannelManager,官方注解:Manages a set of channels, indexed by channel number (1… _channelMax)。

ChannelManager类的代码量不是很多,主要用来管理Channel的,channelNumber=0的除外,应为channelNumber=0是留给Connection的特殊的channelNumber。

下面是ChannelManager的成员变量:

/** Monitor for _channelMap and channelNumberAllocator */
private final Object monitor = new Object();/** Mapping from 1.._channelMax to {@link ChannelN} instance */private final Map _channelMap = new HashMap();private final IntAllocator channelNumberAllocator;private final ConsumerWorkService workService;private final Set shutdownSet = new HashSet();/** Maximum channel number available on this connection. */
private final int _channelMax;
private final ThreadFactory threadFactory;

这上面的成员变量下面会有涉及。


对于ChannelManager的使用,是AMQConnection中的成员变量:

/** Object that manages a set of channels */
private volatile ChannelManager _channelManager;

AMQConnection中start()的_channelManager中对其初始化:

protected ChannelManager instantiateChannelManager(int channelMax, ThreadFactory threadFactory) {return new ChannelManager(this._workService, channelMax, threadFactory);
}

再调用其构造函数:

public ChannelManager(ConsumerWorkService workService, int channelMax, ThreadFactory threadFactory) {if (channelMax == 0) {// The framing encoding only allows for unsigned 16-bit integers// for the channel numberchannelMax = (1 << 16) - 1;}_channelMax = channelMax;channelNumberAllocator = new IntAllocator(1, channelMax);this.workService = workService;this.threadFactory = threadFactory;
}

这里的ConsumerWorkService也在AMQConnection的start()方法中初始化——initializeConsumerWorkService():

private void initializeConsumerWorkService() {this._workService  = new ConsumerWorkService(executor, threadFactory, shutdownTimeout);
}

再回到构造函数。

channelMax参数是在client接收到broker的Connection.Tune帧中的“Channel-Max”参数之后设置的,如果为0则表示没有限制,这里就会设置为默认的最大值:2的16次方-1。
threadFactory参数是指:Executors.defaultThreadFactory();

关于ConsumerWorkService请参考文章末尾处。


使用过RabbitMQ的同学知道要生产或者消费消息之前必须要初始化Channel,如下:

Channel channel = connection.createChannel();

这个createChannel()是AMQConnection中的方法:

public Channel createChannel(int channelNumber) throws IOException {ensureIsOpen();ChannelManager cm = _channelManager;if (cm == null) return null;return cm.createChannel(this, channelNumber);
}
public Channel createChannel() throws IOException {ensureIsOpen();ChannelManager cm = _channelManager;if (cm == null) return null;return cm.createChannel(this);
}

这里就是调用了ChannelManager的createChannel方法。

下面是ChannelManager中关于创建Channel的代码:

public ChannelN createChannel(AMQConnection connection) throws IOException {ChannelN ch;synchronized (this.monitor) {int channelNumber = channelNumberAllocator.allocate();if (channelNumber == -1) {return null;} else {ch = addNewChannel(connection, channelNumber);}}ch.open(); // now that it's been safely addedreturn ch;
}public ChannelN createChannel(AMQConnection connection, int channelNumber) throws IOException {ChannelN ch;synchronized (this.monitor) {if (channelNumberAllocator.reserve(channelNumber)) {ch = addNewChannel(connection, channelNumber);} else {return null;}}ch.open(); // now that it's been safely addedreturn ch;
}private ChannelN addNewChannel(AMQConnection connection, int channelNumber) throws IOException {if (_channelMap.containsKey(channelNumber)) {// That number's already allocated! Can't do it// This should never happen unless something has gone// badly wrong with our implementation.throw new IllegalStateException("We have attempted to "+ "create a channel with a number that is already in "+ "use. This should never happen. "+ "Please report this as a bug.");}ChannelN ch = instantiateChannel(connection, channelNumber, this.workService);_channelMap.put(ch.getChannelNumber(), ch);return ch;
}protected ChannelN instantiateChannel(AMQConnection connection, int channelNumber, ConsumerWorkService workService) {return new ChannelN(connection, channelNumber, workService);
}

上面有两个createChannel方法,一个是带了channelNumber的,一个是自动分片channelNumber的,分别对应AMQConnection中的两个方法。最后都调用addNewChannel方法。

注意两个createChannel方法中都有这样一句代码:

ch.open();

这个是什么呢?其实是调用ChannelN的open方法:

/*** Package method: open the channel.* This is only called from {@link ChannelManager}.* @throws IOException if any problem is encountered*/
public void open() throws IOException {// wait for the Channel.OpenOk response, and ignore itexnWrappingRpc(new Channel.Open(UNSPECIFIED_OUT_OF_BAND));
}

这样就调用了AMQChannel的rpc方法,向broker发送了一个Channel.Open帧。

addNewChannel方法实际上是创建了一个ChannelN对象,然后置其于ChannelManager中的_channelMap中,方便管理。

channelNumberAllocator是channelNumber的分配器,其原理是采用BitSet来实现channelNumber的分配,有兴趣的同学可以深究进去看看。

关于ChannelN类会有专门一篇博文来讲述,其实整个RabbitMQ-client的代码最关键的就是ChannelN这个类,需要着重讲述。

细心的朋友可能会发现关于ConsumerWorkService这个,我并没有做什么阐述。这个主要牵涉到Channel层面的处理,涉及到的类有AMQConnection, ChannelN, ConsumerDispatcher等。ConsumerWorkService是在AMQConnection中初始化,在ChannelManager中引用。至于这里怎么理解,在ChannelN中这么解释:
service for managing this channel’s consumer callbacks。意思是管理消费回调的服务。
综述,ChannelManager主要用来管理Channel, 包括channelNumber与Channel之间的映射关系。

相关内容

热门资讯

喜欢穿一身黑的男生性格(喜欢穿... 今天百科达人给各位分享喜欢穿一身黑的男生性格的知识,其中也会对喜欢穿一身黑衣服的男人人好相处吗进行解...
发春是什么意思(思春和发春是什... 本篇文章极速百科给大家谈谈发春是什么意思,以及思春和发春是什么意思对应的知识点,希望对各位有所帮助,...
网络用语zl是什么意思(zl是... 今天给各位分享网络用语zl是什么意思的知识,其中也会对zl是啥意思是什么网络用语进行解释,如果能碰巧...
为什么酷狗音乐自己唱的歌不能下... 本篇文章极速百科小编给大家谈谈为什么酷狗音乐自己唱的歌不能下载到本地?,以及为什么酷狗下载的歌曲不是...
华为下载未安装的文件去哪找(华... 今天百科达人给各位分享华为下载未安装的文件去哪找的知识,其中也会对华为下载未安装的文件去哪找到进行解...
家里可以做假山养金鱼吗(假山能... 今天百科达人给各位分享家里可以做假山养金鱼吗的知识,其中也会对假山能放鱼缸里吗进行解释,如果能碰巧解...
四分五裂是什么生肖什么动物(四... 本篇文章极速百科小编给大家谈谈四分五裂是什么生肖什么动物,以及四分五裂打一生肖是什么对应的知识点,希...
怎么往应用助手里添加应用(应用... 今天百科达人给各位分享怎么往应用助手里添加应用的知识,其中也会对应用助手怎么添加微信进行解释,如果能...
美团联名卡审核成功待激活(美团... 今天百科达人给各位分享美团联名卡审核成功待激活的知识,其中也会对美团联名卡审核未通过进行解释,如果能...
一帆风顺二龙腾飞三阳开泰祝福语... 本篇文章极速百科给大家谈谈一帆风顺二龙腾飞三阳开泰祝福语,以及一帆风顺二龙腾飞三阳开泰祝福语结婚对应...