序 本文主要研究一下rocketmq-client-go的QueueSelector QueueSelector rocketmq-client-go-v2.0.0/producer/selector.go {代码...} QueueSelector接口，定义了Select方法 manualQueueSelector rocketmq-client-go-v2.0.0/producer/selector.go {代码...} manualQueueSelector的select方法直接返回message.Queue NewRand...

序 本文主要研究一下rocketmq的AsyncAppender AsyncAppender org/apache/rocketmq/logging/inner/LoggingBuilder.java {代码...} 初始化Dispatcher，构造器调用Dispatcher的start，然后close方法调用dispatcher.join() append方法会判断buffer够不够，够的话往buffer添加事件，不够则丢弃同时进行DiscardSummary统计 dis...

这里重写了subAppend方法，先检查当前时间是否大于nextCheck值，大于则更新nextCheck值，然后进行rollOver操作，最后在调用父类的subAppend方法

序 本文主要研究一下rocketmq的RollingFileAppender RollingFileAppender org/apache/rocketmq/logging/inner/LoggingBuilder.java {代码...} 这里重写了subAppend方法，调用父类subAppend方法之后，判断是否需要rollOver 这里定义了maxFileSize，即单个文件的大小，然后还定义了nextRollover索引 这里使用的是CountingQ...

序 本文主要研究一下rocketmq的FileAppender WriterAppender org/apache/rocketmq/logging/inner/LoggingBuilder.java {代码...} 这个接口定义了writeHeader、writeFooter、append等方法 append方法这里参数是LoggingEvent，内部委托给subAppend subAppend方法调用layout进行格式化event，然后如果需要flush，则会直接对...

继承自checked exception，底下有RemotingCommandException、RemotingConnectException、RemotingSendRequestException、RemotingTimeoutException、RemotingTooMuchRequestException

序 本文主要研究一下rocketmq的NettyEncoder及NettyDecoder NettyEncoder org/apache/rocketmq/remoting/netty/NettyEncoder.java {代码...} 这里继承MessageToByteEncoder，类型是RemotingCommand，先写入header再写入body RemotingCommand.encodeHeader org/apache/rocketmq/remoting/protocol/RemotingCommand.java {...

序 本文主要研究一下rocketmq的RequestTask RequestTask org/apache/rocketmq/remoting/netty/RequestTask.java {代码...} 里头有一个runnable、channel以及remotingCommand NettyRemotingAbstract.processRequestCommand org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java {代码...} 这里根据上下文准备...

序 本文主要研究一下rocketmq的NettyClientConfig NettyClientConfig org/apache/rocketmq/remoting/netty/NettyClientConfig.java {代码...} 这里主要有几个参数： clientWorkerThreads，默认为4 clientCallbackExecutorThreads，默认为cpu核数 clientOnewaySemaphoreValue，默认为NettySystemConfig.CLIENT_ONEWAY_SEM...

实现了ChannelEventListener接口，除了onChannelConnect外，其余各个方法均委托给namesrvController的routeInfoManager的onChannelDestroy方法

这里使用HashMap，然后通过ReentrantReadWriteLock进行并发控制，map的key是namespace，而value是一个HashMap

序 本文主要研究一下rocketmq的PushConsumerImpl PushConsumerImpl io/openmessaging/rocketmq/consumer/PushConsumerImpl.java {代码...} 这里创建的是DefaultMQPushConsumer，同时设置的messageListener为MessageListenerImpl attachQueue的时候，使用的是rocketmqPushConsumer.subscribe 启动时调用rocketmqPushConsu...

LocalMessageCache是从consumeRequestCache这个LinkedBlockingQueue中poll出来ConsumeRequest，该request携带了MessageExt

序 本文主要研究一下rocketmq的SequenceProducerImpl SequenceProducerImpl io/openmessaging/rocketmq/producer/SequenceProducerImpl.java {代码...} 采用的是LinkedBlockingQueue，send方法实际调用的是添加到队列 另外提供了commit以及rollback方法，都加了synchronized保证对LinkedBlockingQueue操作的线程安全 com...

另外调用OMSUtil.msgConvert将api的BytesMessage转换为org.apache.rocketmq.common.message.Message

ServiceLifecycle的startup里头调用DefaultMQProducer的start方法，shutdown里头调用DefaultMQProducer的shutdown方法

openmessaging-java/openmessaging-api-samples/src/main/java/io/openmessaging/samples/consumer/PullConsumerApp.java

openmessaging-java/openmessaging-api/src/main/java/io/openmessaging/MessagingAccessPoint.java

openmessaging-java/openmessaging-api/src/main/java/io/openmessaging/producer/Producer.java

namespace，类似cgroup的namespace，用来进行安全隔离，每个namespace有自己的producer、consumer、topic、queue等

redis系列 redis的发布订阅功能 redis消息队列 redis的pipeline redis的scan操作 序 本文主要简述redis作为消息队列的利弊和场景。 消息中间件要素 消息模型 传送模型(发布订阅or点对点)是基于topic的发布订阅还是点对点模型 接收模型(推or拉)消息是由生产者推还是由消费者拉 消息投递可靠性 at-most-once(消息丢失) 消...

主要概念 Exchange Bindings Queues 与JMS的区别 JMS虽然有抽象出Destination，但是具体编程的时候，需要使用者指定是Queue还是Topic。 AMQP，通过Exchange屏蔽了queue和topic的细节。 消息路由 简单的说就是借助Exchange通过routingKey和binding两个的组合使用实现message与queue的路由。 仅仅使用routingKey，不需要ex...

消息中间件，Message-Oriented Middleware，简称MOM。采用消息中间件的作用一般有两点：一是解耦，二是异步（起到削峰填谷的作用）