Day14 RocketmQ 客户端模型编程

Screenshot 2024-12-14 at 00.54.07.png

• 消息确认ACK机制
  FLUAH_DISH ASYNC_FLUSH 等机制.

• 广播消息，TAG标记过滤消息，顺序消息

• batch comsume 批量消息

• 事务消息

• 消费者端消息回调
  ConsumeConcurrently.CONSUME_SUCCESS 或者是LATER.

• 发送方式，同步，异步,单向等落盘策略.

  defaultMQProducer.send(new Message(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                SendStatus sendStatus = sendResult.getSendStatus();
                SEND_OK,
                        FLUSH_DISK_TIMEOUT,
                        FLUSH_SLAVE_TIMEOUT,
                        SLAVE_NOT_AVAILABLE;
            }

            @Override
            public void onException(Throwable throwable) {

            }
        });
        defaultMQProducer.send();
        
        defaultMQProducer.sendOneway();

-- 消费者端

defaultMQPullConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                ConsumeConcurrentlyStatus.RECONSUME_LATER;
                return null;
            }
        });

---

• 消息确认机制:
  Producer: 多次重试+消息确认机制保证。

• one way 单向

• 同步

• 异步发送

---

消费模式:

广播和集群(默认是集群模式)

• 集群消费模式

Screenshot 2024-12-14 at 01.15.21.png

集群可以负载均衡，这样一条数据处理完到消费者就结束了。

• 广播模式: (每台机器都会消费一次)
  比如更新本地缓存. 每台实例都可以收到消息.

---

过滤消息:

Tag: 比如货物出库，入库.

new Message("topic","tagA","message".getBytes());

也可以自定义参数属性进行过滤.

顺序消息

下单顺序保证 :
MessageListenerOrderly接口
下单->锁库存->支付->物流等步骤是有顺序的。 可以借助MQ顺序消息实现.

• 延迟消息:
  18个默认级别

Screenshot 2024-12-14 at 01.21.53.png

改topic名称，然后定时扫描

Screenshot 2024-12-14 at 01.22.34.png

• 批量消息 建议不超过1M.

    SendResult send(final Collection<Message> msgs) throws MQClientException, RemotingException, MQBrokerException,
        InterruptedException;

Screenshot 2024-12-14 at 01.26.11.png

事务消息 2阶段提交

2PC

Screenshot 2024-12-14 at 01.25.34.png Screenshot 2024-12-14 at 01.25.55.png

ACL 权限控制，方便开人。 分配特定topic即可。

• 消息ID 是唯一的，每条收到的msg消息的Id。

• msg Id

但是需要注意，对于客户端来说，毕竟是不知道这个msgId是如何产生的。实际上，在RocketMQ内部，也会针对批量消息、事务消息等特殊的
消息机制，有特殊的msgId分配机制。因此，在复杂业务场景下，不建议使用msgId来作为消息的唯一索引，而建议采用下面的key属性，自行
指定业务层面上的唯一索引

 new Message().setKeys("业务的唯一keyid，作为补充");

对于复杂的业务场景，rocketMQ内部会针对批量消息，事务消息等对msgId做特殊分配，因为需要用keys 补充唯一id依据.

重试16次

Screenshot 2024-12-14 at 01.31.25.png

16次未成功，就投入到死信队列里.

私信队列

• group: %DLQ%+ConsumeGroup

Screenshot 2024-12-14 at 01.33.16.png

死信消息处理需要设置权限，默认是禁读的.
保存的默认时间是3天

比如对于死信做处理，然后发送到正常的topic处理，或者消息.