《RocketMq》四、Client Producer/Consumer总览
flypeng 发布时间:08-08 来源:0 浏览:0次

一、总览

Rocketmq的客户端包括Producer和Consumer,这2个部分。虽然两个部分的细节不同,但在发送消息,维持broker/namesrv连接,等方面有着相同的处理方式,可以对二者统一进行管理。
让我们自己先思考下,一个client需要完成哪些内容:

获取namesrv的地址

与namesrv交互,获取broker的地址以及TopicRouteData

管理进程中所有的producer和consumer

与broker/namesrv如何交互,长链接or短链接

pull消息

send消息

负载均衡

这里来对应rocketmq分析一下他是如何实现上述功能的:
1. PullAPIWrapper:主要用于上述5,pull消息,以及接收到消息之后的tags过滤;
2. RebalanceImpl:用于上述的7,负载均衡;
3. MQClientInstance:
A. MQClientAPIImpl:真实的调用remotingserver进行消息的发送和接收,涉及到与broker,namesrv,的各类交互,通讯实现的包裹类;
B. PullMessageService:这个应该只有在consumer push模式使用,因为push其实是后台包装了一个不断pull的请求实现的,所以会把pullrequest放入pullRequestQueue中,不断的消费拉取执行;
C. RebalanceService:负载均衡策略,对所有的consumer调用doRebalance()进行mq与consumer的重分配;
D. DefaultMQProducerImpl:产生一个CLIENT_INNER_PRODUCER,在消息消费失败的时候重新放入mq中;

设计细节

MQClientManager:单例类,用于管理MQClientInstance,详见

ConcurrentMap<String/* clientId */, MQClientInstance> factoryTable;

其中的clientId为IP@pid,也可以通过instanceName设置为IP@instanceName; 也就是说所有相同clientId的instance都在factoryTable里面管理;
2. MQClientInstance:用于管理所有的consumer,producer以及该实例的路由信息,remotingServer等

private final ClientConfig clientConfig;
private final int instanceIndex;
private final String clientId;
private final long bootTimestamp = System.currentTimeMillis();
private final ConcurrentMap<String/* group */, MQProducerInner> producerTable = new ConcurrentHashMap<String, MQProducerInner>();
private final ConcurrentMap<String/* group */, MQConsumerInner> consumerTable = new ConcurrentHashMap<String, MQConsumerInner>();
private final ConcurrentMap<String/* group */, MQAdminExtInner> adminExtTable = new ConcurrentHashMap<String, MQAdminExtInner>();
private final NettyClientConfig nettyClientConfig;
private final MQClientAPIImpl mQClientAPIImpl; // 发送的具体实现
private final MQAdminImpl mQAdminImpl;
private final ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<String, TopicRouteData>();
private final Lock lockNamesrv = new ReentrantLock();
private final Lock lockHeartbeat = new ReentrantLock();
private final ConcurrentMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>> brokerAddrTable =
new ConcurrentHashMap<String, HashMap<Long, String>>();
private final ConcurrentMap<String/* Broker Name */, HashMap<String/* address */, Integer>> brokerVersionTable =
new ConcurrentHashMap<String, HashMap<String, Integer>>();
private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "MQClientFactoryScheduledThread");
}
});
private final ClientRemotingProcessor clientRemotingProcessor;
private final PullMessageService pullMessageService; // 拉取消息
private final RebalanceService rebalanceService; // 负载均衡
private final DefaultMQProducer defaultMQProducer;
private final ConsumerStatsManager consumerStatsManager;
private final AtomicLong sendHeartbeatTimesTotal = new AtomicLong(0);
private ServiceState serviceState = ServiceState.CREATE_JUST;
private DatagramSocket datagramSocket;
private Random random = new Random();


3. OffsetStore:分为LocalFileOffsetStore,用于将当前消费到的位置保存到本地文件;RemoteBrokerOffsetStore ,用于将当前消费到的位置保存到broker;
4. PullMessageService(后台任务)

private final LinkedBlockingQueue<PullRequest> pullRequestQueue = new LinkedBlockingQueue<PullRequest>();

rocketmq中其实没有真正的push模式,push模式是客户端在后台不断的pull模拟而来的。对于push模式,其采用的通讯策略都是async,拉取消息成功后调用PullCallback,将消息存入processQueue中,以便后续消费。在push模式下,每次pull消息成功后的pullback中,都会放入下一个request。而什么时候首次放入request,那就是rebalanceImpl后本MQClientInstance新分配的mq就会被加入到PullRequest中,发起第一次的PullRequest;
6. 负载均衡
Producer: MQFaultStrategy
Consumer: RebalanceImpl

protected final ConcurrentMap<String/* topic */, Set<MessageQueue>> topicSubscribeInfoTable =
new ConcurrentHashMap<String, Set<MessageQueue>>();
protected final ConcurrentMap<String /* topic */, SubscriptionData> subscriptionInner =
new ConcurrentHashMap<String, SubscriptionData>();


7. ConsumerMessageService: 这个是在pull消息成功后,消费消息的策略。这个和用户注册的listener种类有关系。它也是consumer中使用的,分为2类:
ConsumeMessageConcurrentlyService:对应客户端的MessageListenerConcurrently, 这个部分是接收到消息之后立即消费;
ConsumeMessageOrderlyService:对应客户端的MessageListenerOrderly,这个部分是接收到消息之后顺序消费;

如果你有好的win10资讯或者win10教程,以及win10相关的问题想要获得win10系统下载的关注与报道。
欢迎加入发送邮件到657025171#qq.com(#替换为@)。期待你的好消息!