RocketMQ-创新互联

• 引入依赖
• 发送消息
• 消费消息
• 消息的可靠性（不丢失、高可用）
• 延时消息
• 消息过滤
• 顺序消息
• • 全局顺序消息
  • 部分顺序消息
• 消息积压
• 消息重复（保证消息的幂等性）
• 消息 重试
• 事务消息
• 问题
• • MQ集群挂掉

org.springframework.bootspring-boot-starter-weborg.apache.rocketmqrocketmq-client4.9.3

// 同步发送
SendResult sendResult = producer.send(msg);
// 异步发送，指定回调
producer.send(msg, new SendCallback() {// 当producer接收到MQ发送来的ACK后就会触发该回调方法的执行
    @Override
    public void onSuccess(SendResult sendResult) {System.out.println(sendResult);
    }
    @Override
    public void onException(Throwable e) {e.printStackTrace();
    }
});

// 消息发送的状态
public enum SendStatus {SEND_OK,       // 发送成功
    FLUSH_DISK_TIMEOUT,  // 刷盘超时。当Broker设置的刷盘策略为同步刷盘时才可能出现这种异常状态。异步刷盘不会出现
    FLUSH_SLAVE_TIMEOUT, // Slave同步超时。当Broker集群设置的Master-Slave的复制方式为同步复制时才可能出现这种异常状态。异步复制不会出现
    SLAVE_NOT_AVAILABLE, // 没有可用的Slave。当Broker集群设置为Master-Slave的复制方式为同步复制时才可能出现这种异常状态。异步复制不会出现
}

DefaultMQPushConsumer consumer = new
        DefaultMQPushConsumer ("cg");
consumer.setNamesrvAddr ("rocketmqOS:9876");
// 指定 从第一条消息开始消费
consumer.setConsumeFromWhere (ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe ("someTopicA", "*");

// 指定 每次可以 消费10条消息，默认为1
consumer.setConsumeMessageBatchMaxSize (10);
// 指定 每次 可以从Broker拉取40条消息， 默认为32
consumer.setPullBatchSize (40);
consumer.registerMessageListener (new MessageListenerConcurrently () {@Override
    public ConsumeConcurrentlyStatus consumeMessage(Listmsgs, ConsumeConcurrentlyContext context) {for (MessageExt msg : msgs) {System.out.println (msg);
        }
        // 消费成功的返回结果
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        // 消费异常时的返回结果
        // return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }
});
consumer.start ();

消息可能在哪些阶段丢失呢？可能会在这三个阶段发生丢失：生产阶段、存储阶段、消费阶段

生产阶段：事务消息，写到OS Cache就返回 成功发送！

存储阶段：调整 MQ的刷盘策略，我们需要调整broker.conf配置文件，将其中的flushDiskType配置设置为：SYNC_FLUSH，默认他的值是ASYNC_FLUSH，即默认是异步刷盘

消费阶段

不能 异步 消费消息

// 开启监听，用于接收消息
consumer.registerMessageListener (new MessageListenerConcurrently () {@Override
    public ConsumeConcurrentlyStatus consumeMessage(Listlist, ConsumeConcurrentlyContext consumeConcurrentlyContext) {// 业务 处理
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});
// 启动接收消息的服务
consumer.start ();

也可以 使用如下结构：

目前RocketMQ支持的延时级别：

private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

生产者：

// 这是 订单系统的生产者
DefaultMQProducer producer = new DefaultMQProducer ("OrderGroup");
// 启动生产者
producer.start ();
Message msg = new Message (
        "CreateOrderInformTopic",
        "create success".getBytes ());
// 设置消息为延时消息，延迟级别为 16
msg.setDelayTimeLevel (16);
producer.send (msg);

消费者：

// 订单扫描服务 的消费者
DefaultMQPushConsumer consumer =
        new DefaultMQPushConsumer ("OrderScanServiceGroup");
// 订阅 订单创建 通知Topic
consumer.subscribe ("CreateOrderInformTopic", "*");
// 注册 消息监听者
consumer.registerMessageListener (new MessageListenerConcurrently () {@Override
    public ConsumeConcurrentlyStatus consumeMessage(Listlist,
                                                    ConsumeConcurrentlyContext consumeConcurrentlyContext) {for (MessageExt msg : list) {// 将 msg转为 订单对象
            Order order = buildOrder (msg);
            // 根据Id 查询数据库
            order = getById (order.getId ());
            // 如果 已支付
            if (order.getStatus () == 1) {}
            // 如果 未支付
            else {}
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

• 根据Tag过滤

// 创建消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer ("aGroup");
// 订阅主题，tag为 a,b,c的都监听
consumer.subscribe ("orderInfoTopic", "a || b || c");

• SQL 表达式过滤

修改broker.conf 配置文件才能生效

// 创建消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer ("bGroup");
// 只有订阅的消息有这个属性a, a >=0 and a<= 3
consumer.subscribe ("TopicTest", MessageSelector.bySql ("a between 0 and 3"));
consumer.registerMessageListener (new MessageListenerConcurrently () {@Override
    public ConsumeConcurrentlyStatus consumeMessage(Listmsgs, ConsumeConcurrentlyContext context) {// 处理业务逻辑
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});
consumer.start ();

RocketMQ还是 支持比较丰富的数据过滤语法的，如下所示：
（1）数值比较，比如：>，>=，<，<=，BETWEEN，=；
（2）字符比较，比如：=，<>，IN；
（3）IS NULL 或者 IS NOT NULL；
（4）逻辑符号 AND，OR，NOT；
（5）数值，比如：123，3.1415；
（6）字符，比如：‘abc’，必须用单引号包裹起来；
（7）NULL，特殊的常量
（8）布尔值，TRUE 或 FALSE

顺序消息是指消息的消费顺序和产生顺序相同，在有些业务逻辑下，必须保证顺序，比如订单的生成、付款、发货，这个消息必须按顺序处理才行。

「消息乱序解决方案 不能和重试队列混用。」

创建一个 Topic ，默认八个写队列，八个读队列

要保证全局顺序消息， 需要先把 Topic 的读写队列数设置为 一，然后Producer Consumer 的并发设置，也要是一

部分顺序消息相对比较好实现，生产端需要做到把同 ID 的消息发送到同一个 Message Queue ；在消费过程中，要做到从同一个Message Queue读取的消息顺序处理

DefaultMQProducer producer = new DefaultMQProducer ();
producer.setSendLatencyFaultEnable (true);
producer.setSendMsgTimeout (50);
producer.setNamesrvAddr ("192.168.111.101:9876");
producer.start ();
Message message = new Message ("Topic", "hello world".getBytes ());
SendResult sendResult = producer.send (message, new MessageQueueSelector () {@Override
            public MessageQueue select(Listmqs,
                                       Message message,
                                       Object arg) {Long orderId = (Long) arg; // 根据 订单id选择发送 queue
                long index = orderId % mqs.size (); // 用订单id 对MessageQueue数量取模
                return mqs.get ((int) index);  // 返回一个MessageQueue
            }
        }, orderId //这里传入订单id
);

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer ("cg");
consumer.setNamesrvAddr ("192.168.111.101:9876");
// 从头 开始消费
consumer.setConsumeFromWhere (ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe ("Topic", "*");
consumer.registerMessageListener (new MessageListenerOrderly () {@Override
  public ConsumeOrderlyStatus consumeMessage(Listmsgs,
                                             ConsumeOrderlyContext context) {  context.setAutoCommit (true);
      try {  for (MessageExt msg : msgs) {  // 对有序的消息 进行处理
          }
          return ConsumeOrderlyStatus.SUCCESS;
      } catch (Exception e) {  // 如果 消息处理 有问题
          // 让这批消息 暂停一会儿 再继续处理这批消息
          return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
      }
  }
});
consumer.start ();

发生了消息积压，这时候就得想办法赶紧把积压的消息消费完，就得考虑提高消费能力，一般有两种办法

• 消费者扩容：如果当前Topic的Message Queue的数量大于消费者数量，就可以对消费者进行扩容，增加消费者，来提高消费能力，尽快把积压的消息消费玩。
• 消息迁移Queue扩容：如果当前Topic的Message Queue的数量小于或者等于消费者数量，这种情况，再扩容消费者就没什么用，就得考虑扩容Message Queue。可以新建一个临时的Topic，临时的Topic多设置一些Message Queue，然后先用一些消费者把消费的数据丢到临时的Topic，因为不用业务处理，只是转发一下消息，还是很快的。接下来用扩容的消费者去消费新的Topic里的数据，消费完了之后，恢复原状
  

处理消息重复问题，主要有业务端自己保证，主要的方式有两种：业务去重 和 消息去重。

• 业务去重：保证 业务消费的幂等性即可

• 消息去重：依据在 生产者方设置消息的 messageKey，然后 每一条消息 在消费方依据这个唯一的 messageKey，进行幂等判断
  流程：** insert 失败，说明已经消费---->捕获异常返回已消费，insert 成功---->处理业务 提交事务后 再确认成功消费~~ **
  

// 设置 同步发送失败时  重试发送的次数，默认为2次
producer.setRetryTimesWhenSendFailed(3);
// 设置 发送 超时时限为5s，默认3s
producer.setSendMsgTimeout(5000);
// 指定 异步发送失败后 不进行重试发送
producer.setRetryTimesWhenSendAsyncFailed(0);



// 顺序消息消费失败的消费重试时间间隔，单位毫秒，默认为1000，其取值范围为[10,30000]
// 重试期间 应用会出现 消息消费被阻塞
consumer.setSuspendCurrentQueueTimeMillis(100);
// 修改消费重试次数,默认16次
consumer.setMaxReconsumeTimes(10);

消息 刷盘失败策略

消息刷盘超时（Master或Slave）或slave不可用（slave在做数据同步时向master返回状态不是SEND_OK）时，默认是不会将消息尝试发送到其他Broker的。不过，对于重要消息可以通过在Broker的配置文件设置retryAnotherBrokerWhenNotStoreOK属性为true来开启

重试队列里面的消息会再次发给消费组，默认 最多重试 16 次，如果重试 16 次失败则进入「死信队列」

「死信队列：」
对于死信队列，一般我们可以专门开一个后台线程，订阅这个死信队列，对死信队列中的消息，一直不停的尝试。

消息量比较大，不建议同步，影响消息消费速度，造成消息堆积

消费者处理失败后，立刻写入重试表，有个 定时任务专门重试

消费失败，自己给同一个topic发一条消息—>对消息顺序要求不高的场景可以使用

原理

Rocketmq 未收到rollback、commit也会 补偿回调，MQ也会有补偿机制 ：checkLocalTransaction方法
让我们自己处理

TransactionListenerImpl 事务监听器

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

public class TransactionListenerImpl implements TransactionListener {// 发送成功给 broker后，可以 执行本地事务
    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {// 执行 订单本地事务
        try {// 如果 本地事务都执行成功了，返回 commit
            return LocalTransactionState.COMMIT_MESSAGE;

        } catch (Exception e) {// 本地事务 执行失败，回滚 本地事务
            // 更新 broker中的消息状态为 删除

            return LocalTransactionState.ROLLBACK_MESSAGE;
        }

    }

    // 因为超时 等原因，没有返回 commit或者 rollback
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {// 这里默认是 回滚
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
}

生产者：

// 接受 RocketMQ 回调的一个 监听器接口
// 会执行 订单本地事务，commit、rollback，回调查询 等逻辑
TransactionListenerImpl transactionListener = new TransactionListenerImpl ();
TransactionMQProducer producer = new TransactionMQProducer ();
ThreadPoolExecutor executorService = new ThreadPoolExecutor (
        2,
        5,
        100,
        TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(2000)
);
// 设置 对应的线程池，负责执行 回调请求
producer.setExecutorService (executorService);
// 开启 容错机制
producer.setSendLatencyFaultEnable (true);
// 设置 事务监听器
producer.setTransactionListener (transactionListener);
// 设置 发送失败时，最多重试几次
producer.setRetryTimesWhenSendFailed (2);

// 构建 消息体
Message message = new Message (
        "PayOrderSuccessTopic",
        "Tag",
        "MyKey",
        "Pay Success".getBytes ());
// 可以 查询发送结果
SendResult sendResult = producer.send (message, 10);

降级方案通常的思路：
发送消息到MQ代码里去try catch捕获异常，如果你发现发送消息到MQ有异常，此时你需要进行重试

重试了，比如超过3次还是失败，说明此时可能就是你的MQ集群彻底崩溃了
此时你必须把这条重要的消息写入到本地存储中去，可以是写入数据库里

要不停的尝试发送消息到MQ去
发现MQ集群恢复了，你 必须有一个后台线程可以 把之前持久化存储的消息都查询出来，然后依次 按照顺序 发送到MQ集群里去

这里要有一个很关键的注意点，就是你把消息写入存储中 暂存时，一定要保证他的顺序，比如按照顺序一条一条的写入本地磁盘文件去
暂存消息

**流量 太多：**解决⽅法就是 对线上系统扩容双段缓冲的⼤⼩，从 512kb 扩容到⼀个缓冲区 10mb。

你是否还在寻找稳定的海外服务器提供商？创新互联www.cdcxhl.cn海外机房具备T级流量清洗系统配攻击溯源，准确流量调度确保服务器高可用性，企业级服务器适合批量采购，新人活动首月15元起，快前往官网查看详情吧