RocketMQ-生产者核心解析、主从同步机制解析,生产者同步异步消息发送

    • 生产者核心参数
    • Master – Slave主从同步机制解析
      • 数据内容同步
      • 元数据信息同步
      • 通信协议
    • 生产者消息发送
      • 生产者同步消息发送
      • 生产者异步消息发送

生产者核心参数


producerGroup: 组名
createTopicKey:创建topic,实际生产实践不允许生产者创建top。
defaultTopicQueueNums(默认为4):默认的topic关联的队列数量
sendMsgTimeout(单位:ms):发送消息连接broker超时时间。
compressMsgBodyOverHowmuch(默认压缩字节4096):消息体达到多少压缩。
retryTimesWhenSendFailed (可配置):发送失败重试次数
retryAnotherBrokerWhenNotStoreOK(默认false):发送broker存储失败换个broker发送。
maxMessageSize(默认128K):消息最大可以设置多大。
heartbeatBrokerInterval:与broker的心跳间隔(以微秒为单位,默认为30毫秒)

Master – Slave主从同步机制解析

同步的信息主要是数据内容和元数据信息

数据内容同步

实时进行同步,同步的是commitlog中的数据,对实时性要求高,并且丢失数据就无法恢复。使用原生socket进行同步。
源码分析
数据内容同步主要是在rocketmq-store中。主要涉及 HAConnectionHAServiceWaitNotifyObject三个类。并没有使用netty而是原生nio,是为了更加高效。

  • 对于Master节点

    HAService
    AcceptSocketService(内部类):接受slave节点连接。

    HAConnection
    ReadSocketService(内部类):读来自Slave节点的数据。
    WriteSocketService(内部类):写往到Slave节点的数据。

  • 对于Slave节点

HAService
HAClient(内部类):对Master节点连接、读写数据。

元数据信息同步

broker判断如果是slave节点,那么会启动定时任务不断同步,如果丢失也可以从其他地方重试获取。包含topic信息和offset等。使用netty同步。
源码分析
元数据同步主要发生在broker,所以这部分代码主要在rocketmq-broker的模块中。是在handleSlaveSynchronize方法中通过定义了一个固定时间的定时任务,时间是10秒钟执行一次,当然前提条件是broker节点的角色是slave,而broker节点是master时,如果有定时任务会取消,因为master是不用同步元数据信息。

这个方法会有三个地方调用:

  • broker刚刚启动
  • master切换成slave
  • slave切换成master。

定时任务的逻辑是写在syncAll方法中。主要是需要同步4部分内容:

  • 同步topic配置信息
  • 同步消费者偏移量
  • 同步延时偏移量
  • 同步订阅组配置信息。


4个方法的内容实际上就是封装了netty做rpc的调用,对不同的操作都会对应到一个code。


通信协议

Slave => Master:上报CommitLog已经同步到的物理位置。使用maxPhyOffset字段,代表CommitLog接受到的最大物理位置。
Master => Slave:传输新的CommitLog数据。使用fromPhyOffset字段,代表CommitLog开始传输的物理位置。

生产者消息发送

生产者同步消息发送

消息的同步发送:producer.send(msg)
同步发送消息核心实现:DefaultMQProducerlmpl
同步发送消息可以直接获取返回值:

// 同步发送消息,直接获取发送结果SendResult sr = producer.send(message, new MessageQueueSelector() {    @Override    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {        Integer queueNumber = (Integer)arg;        return mqs.get(queueNumber);    }}, 2);System.err.println(sr);

生产者异步消息发送

producer.send(Message msg, SendCallback sendCallback)
异步发送消息核心实现:DefaultMQProducerlmpl
异步发送消息需要通过回调函数获取返回值:

// 异步发送消息,回调函数获取结果producer.send(message, new SendCallback() {    // 可靠性消息投递    @Override    public void onSuccess(SendResult sendResult) {        System.err.println("msgId: " + sendResult.getMsgId() + ", status: " + sendResult.getSendStatus());    }    @Override    public void onException(Throwable e) {        // 发送失败做异常处理        e.printStackTrace();        System.err.println("------发送失败");    }});

本文内容到此结束了,
如有收获欢迎点赞收藏关注✔️,您的鼓励是我最大的动力。
如有错误❌疑问欢迎各位大佬指出。
主页