[Java知识库]发消息就用消息队列——kafka、rocketMQ(二)

rocketMQ介绍

环境搭建

• 下载rocketMQ

• 解压下载的rocketMQ

upzip rocketmq-all-4.9.2-bin-release.zip

• 调整启动的内存，不然启动就会报错

编辑：vim runserver.sh

编辑：vim runbroker.sh

如果使用docker进行搭建的话 还需要要修改一个配置broker.conf

• 启动

#启动nameserv
sh mqnamesrv
# 启动mq 服务
sh mqbroker -n localhost:9876  -c ../conf/broker.conf

• 查看是否启动成功

tail -f ~/logs/rocketmqlogs/broker.log
# 出现下图就说明已经启动成功了

基本使用

1. 环境搭建

 <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.5.2</version>
</dependency>

2. 基本使用

• 负载均衡

#发送
public static void main(String[] args) throws Exception {
        //1.创建一个发送消息的对象Producer
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.设定发送的命名服务器地址
        producer.setNamesrvAddr("10.199.12.155:9876");
        //3.1启动发送的服务
        producer.start();
        //4.创建要发送的消息对象,指定topic，指定内容body
        Message msg = new Message("topic1","hello rocketmq".getBytes("UTF-8"));
        //4.2发送消息
        SendResult result = producer.send(msg);
        System.out.println("返回结果："+result);
        //5.关闭连接
        producer.shutdown();
}

#消费者
public static void main(String[] args) throws Exception {
        //1.创建一个接收消息的对象Consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.设定接收的命名服务器地址
        consumer.setNamesrvAddr("10.199.12.155:9876");
        //3.设置接收消息对应的topic,对应的sub标签为任意*
        consumer.subscribe("topic1","*");
        //3.开启监听，用于接收消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //遍历消息
                for(MessageExt msg : list){
//                    System.out.println("收到消息："+msg);
                    System.out.println("消息："+new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功处理, mq 收到这个 标记后相同的消息讲不会再次发给消费者
            }
        });
        //4.启动接收消息的服务
        consumer.start();// 开启多线程 监控消息,持续运行
        System.out.println("接收消息服务已开启运行");
    }

• 广播模式

# 生产者 跟上面一样
# 消费者
//1.创建一个接收消息的对象Consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        System.out.println(consumer.getInstanceName());
        //consumer.setInstanceName("instance01");
        //2.设定接收的命名服务器地址
        consumer.setNamesrvAddr("10.199.12.155:9876");
        //3.设置接收消息对应的topic,对应的sub标签为任意*
        consumer.subscribe("topic1","*");

        //设置当前消费者的消费模式（默认模式：负载均衡）
       // consumer.setMessageModel(MessageModel.CLUSTERING);
        //设置当前消费者的消费模式为广播模式：所有客户端接收的消息都是一样的
         consumer.setMessageModel(MessageModel.BROADCASTING);

        //3.开启监听，用于接收消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //遍历消息
                for(MessageExt msg : list){
//                  System.out.println("收到消息："+msg);
                    System.out.println("消费者1："+new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //4.启动接收消息的服务
        consumer.start();
        System.out.println("接收消息服务已开启运行");

广播模式的特点：

1. 如果生产者先发送消息，后启动消费者，消息只能被消费一次。
2. 如果多个消费者先启动（广播模式），后发消息，才有广播效果。
3. 要想有广播模式，必须先启动消费者，然后在启动发送者。

• 消息类型

# 同步消息发送/异步消息发送/单向消息
public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        producer.setNamesrvAddr("10.199.12.155:9876");
        producer.start();
        for (int i = 1; i <= 5; i++) {
            //同步消息发送
            Message msg = new Message("topic2",("同步消息：hello rocketmq "+i).getBytes("UTF-8"));
            SendResult result = producer.send(msg);
            System.out.println("返回结果："+result);


            //异步消息发送
            Message msg2 = new Message("topic2",("异步消息：hello rocketmq "+i).getBytes("UTF-8"));
                producer.send(msg, new SendCallback() {
                    //表示成功返回结果
                    public void onSuccess(SendResult sendResult) {
                        System.out.println(sendResult);
                    }
                    //表示发送消息失败
                    public void onException(Throwable t) {
                        System.out.println(t);
                    }
                });

            //单向消息
            Message msg3 = new Message("topic2",("单向消息：hello rocketmq "+i).getBytes("UTF-8"));
            producer.sendOneway(msg);
        }
        //添加一个休眠操作，确保异步消息返回后能够输出
        TimeUnit.SECONDS.sleep(10);

        producer.shutdown();
    }

• 延时消息

# 延迟发送消息，比如取消订单的时候。
public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        producer.setNamesrvAddr("10.199.12.155:9876");
        producer.start();
        for (int i = 1; i <= 5; i++) {
            Message msg = new Message("topic3",("非延时消息：hello rocketmq "+i).getBytes("UTF-8"));
           // 30秒后再发送,而是先发送,但是通知mq , 30s 才对外暴露数据
            //设置当前消息的延时效果(比如订单,下订单后,20分钟后,决定这个订单是否删除,)
            msg.setDelayTimeLevel(3);
            // 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
            SendResult result = producer.send(msg);
            System.out.println("返回结果："+result);
        }
        producer.shutdown();
    }

• 批量消息

List<Message> msgList = new ArrayList<Message>();
SendResult send = producer.send(msgList);
# 注意点：
消息的总长度不超过4M
消息内容长度包括如下：
	topic（字符串字节数）
	body（字节数组长度）
	消息追加的属性（key与value对应字符串字节数）
    日志（固定20字节）

• Tag

#发送者
public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        producer.setNamesrvAddr("10.199.12.155:9876");
        producer.start();

        //创建消息的时候除了制定topic，还可以指定tag
        Message msg = new Message("topic6","tag2",("消息过滤按照tag：hello rocketmq 2").getBytes("UTF-8"));

        SendResult send = producer.send(msg);
        System.out.println(send);

        producer.shutdown();
    }
#消费者
*代表任意tag
"tag1 || tag2" 代表两个  tag  那个都行
//接收消息的时候，除了制定topic，还可以指定接收的tag,*代表任意tag

consumer.subscribe("topic6","tag1 || tag2");

• 顺序消息

默认情况下，MQ开启了多个队列，同时发送多个消息的话，发送给那个队列是不确定的，同时消息的消费者读取消息，每读取一个消息开启一个线程，也不能保证消息的顺序性。要想保证消息的有序性，需要指定消息的队列，同时消息的消费者应该一个队列开启一个线程进行接收而不是一个消息一个线程

#发送者
   public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        producer.setNamesrvAddr("10.199.12.155:9876");
        producer.start();

        //创建要执行的业务队列
        List<Order> orderList = new ArrayList<Order>();

        Order order11 = new Order();
        order11.setId("a");
        order11.setMsg("主单-1");
        orderList.add(order11);

        Order order12 = new Order();
        order12.setId("a");
        order12.setMsg("子单-2");
        orderList.add(order12);

        Order order13 = new Order();
        order13.setId("a");
        order13.setMsg("支付-3");
        orderList.add(order13);

        Order order14 = new Order();
        order14.setId("a");
        order14.setMsg("推送-4");
        orderList.add(order14);

        Order order21 = new Order();
        order21.setId("b");
        order21.setMsg("主单-1");
        orderList.add(order21);

        Order order22 = new Order();
        order22.setId("b");
        order22.setMsg("子单-2");
        orderList.add(order22);

        Order order31 = new Order();
        order31.setId("c");
        order31.setMsg("主单-1");
        orderList.add(order31);

        Order order32 = new Order();
        order32.setId("c");
        order32.setMsg("子单-2");
        orderList.add(order32);

        Order order33 = new Order();
        order33.setId("c");
        order33.setMsg("支付-3");
        orderList.add(order33);

        //设置消息进入到指定的消息队列中
        for(final Order order : orderList){
            Message msg = new Message("orderTopic",order.toString().getBytes());
            //发送时要指定对应的消息队列选择器
            SendResult result = producer.send(msg, new MessageQueueSelector() {
                //设置当前消息发送时使用哪一个消息队列
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    System.out.println(list.size());// 数量只能通过修改 mq 的配置 改变(阿里开发团队认为,这个是敏感资源需要服务器管理员控制,而不是编码人员控制)
                    //根据发送的信息不同，选择不同的消息队列
                    //根据id来选择一个消息队列的对象，并返回->id得到int值
                    int mqIndex = order.getId().hashCode() % list.size();
                    return list.get(mqIndex);
                }
            }, null);
            System.out.println(result);
        }

        producer.shutdown();
    }
#接收者
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        consumer.setNamesrvAddr("10.199.12.155:9876");
        consumer.subscribe("orderTopic","*");

        //使用单线程的模式从消息队列中取数据，一个线程绑定一个消息队列
        consumer.registerMessageListener(new MessageListenerOrderly() {
            //使用MessageListenerOrderly接口后，对消息队列的处理由一个消息队列多个线程服务，转化为一个消息队列一个线程服务
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                for(MessageExt msg : list){
                    System.out.println(Thread.currentThread().getName()+"  消息："+new String(msg.getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
        System.out.println("接收消息服务已开启运行");
    }

• 事务消息
  MQ的事务流程
  
  MQ 的消息补偿过程(当本地代码执行失败时)
  
  注意点:事务消息仅与生产者有关系，与消费者无关。

#生产者代码
 public static void main1(String[] args) throws Exception {
        //事务消息使用的生产者是TransactionMQProducer
        TransactionMQProducer producer = new TransactionMQProducer("group1");
        producer.setNamesrvAddr("10.199.12.155:9876");
        //添加本地事务对应的监听
        producer.setTransactionListener(new TransactionListener() {
            //正常事务过程
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
               
               // 此处写本地事务处理业务
               // 如果成功,消息改为提交,如果失败改为 回滚,如果是多线程处理状态未知,就提交为未知等待事务补偿过程
               
               //事务提交状态
                return LocalTransactionState.COMMIT_MESSAGE;// 类似于msql 的 commit
            }
            
            //事务补偿过程
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                return null;
            }
        });
        producer.start();

        Message msg = new Message("topic8",("事务消息：hello rocketmq ").getBytes("UTF-8"));
        SendResult result = producer.sendMessageInTransaction(msg,null);
        System.out.println("返回结果："+result);
        producer.shutdown();
    }

# 补偿代码
 public static void main(String[] args) throws Exception {
        //事务消息使用的生产者是TransactionMQProducer
        TransactionMQProducer producer = new TransactionMQProducer("group1");
        producer.setNamesrvAddr("10.199.12.155:9876");
        //添加本地事务对应的监听
        producer.setTransactionListener(new TransactionListener() {
            //正常事务过程
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                //中间状态
                return LocalTransactionState.UNKNOW;
            }
            //事务补偿过程
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("事务补偿过程执行");
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        producer.start();

        Message msg = new Message("topic11",("事务消息：hello rocketmq ").getBytes("UTF-8"));
        SendResult result = producer.sendMessageInTransaction(msg,null);
        System.out.println("返回结果："+result);
        //事务补偿过程必须保障服务器在运行过程中，否则将无法进行正常的事务补偿
//      producer.shutdown();
    }

• rocketMQ高级特性

1. 不用数据库进行存储，而是使用的文件进行存储。
2. 通过启动时初始化文件大小来保证，占用固定的磁盘空间，来保证磁盘读写速度
3. 零拷贝技术，数据传输由传统的4次复制简化为3次复制，减少了一次复制的次数。
4. java语言中使用MappedByteBuffer类实现了该技术
5. 刷盘机制

同步刷盘：
1)生产者发送消息到MQ，MQ接到消息数据
2)MQ挂起生产者发送消息的线程
3)MQ将消息数据写入内存
4)内存数据写入硬盘
5)磁盘存储后返回SUCCESS
6)MQ恢复挂起的生产者线程
7)发送ACK到生产者

异步刷盘：
1)生产者发送消息到MQ，MQ接到消息数据
2)MQ将消息数据写入内存
3)发送ACK到生产者
--等消息量多了--
4)内存数据写入硬盘

优缺点对比：
同步刷盘：安全性高，效率低，速度慢（适用于对数据安全要求较高的业务）
异步刷盘：安全性低，效率高，速度快（适用于对数据处理速度要求较高的业务）

配置方式
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=SYNC_FLUSH


消息服务器：
主从架构（2M-2S） ,即使又一台服务器宕机, 服务依旧可以正常提供
注意: master 一旦宕机,slave 只提供消费服务,不能写入新的消息(slave 不会升级为master)


消息消费：
RocketMQ自身会根据master的压力确认是否由master承担消息读取的功能，当master繁忙时候，自动切换由slave承担数据读取的工作

6. 主从数据复制

同步复制：
master接到消息后，先复制到slave，然后反馈给生产者写操作成功
优点：数据安全，不丢数据，出现故障容易恢复
缺点：影响数据吞吐量，整体性能低
异步复制：
master接到消息后，立即返回给生产者写操作成功，当消息达到一定量后再异步复制到slave
优点：数据吞吐量大，操作延迟低，性能高
缺点：数据不安全，会出现数据丢失的现象，一旦master出现故障，从上次数据同步到故障时间的数据将丢失

配置(配置在启动时 -c 指定的配置文件中  broker.conf)：
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=SYNC_MASTER

7. 负载均衡

Producer负载均衡：内部实现了不同broker集群中对同一topic对应消息队列的负载均衡
Consumer负载均衡：平均分配  循环平均分配 

8. 消息重试

顺序消息重试：
当消费者消费消息失败后，RocketMQ会自动进行消息重试（每次间隔时间为 1 秒）
注意：应用会出现消息消费被阻塞的情况，因此，要对顺序消息的消费情况进行监控，避免阻塞现象的发生
无序消息重试：
无序消息包括普通消息、定时消息、延时消息、事务消息
无序消息重试仅适用于负载均衡（集群）模型下的消息消费，不适用于广播模式下的消息消费
为保障无序消息的消费，MQ设定了合理的消息重试间隔时长

9. 死信队列

概念：
当消息消费重试到达了指定次数（默认16次）后，MQ将无法被正常消费的消息称为死信消息（Dead-Letter Message）
死信消息不会被直接抛弃，而是保存到了一个全新的队列中，该队列称为死信队列（Dead-Letter Queue）
特征：
- 归属某一个组（Gourp Id），而不归属Topic，也不归属消费者
- 一个死信队列中可以包含同一个组下的多个Topic中的死信消息
- 死信队列不会进行默认初始化，当第一个死信出现后，此队列首次初始化

- 不会被再次重复消费
- 死信队列中的消息有效期为3天，达到时限后将被清除


死信消息处理：
在监控平台中，通过查找死信，获取死信的messageId，然后通过id对死信进行精准消费

10. 消息重复消费以及消息幂等性

重复消息原因：
1 生产者发送了重复的消息
    网络闪断
    生产者宕机
2 消息服务器投递了重复的消息
	网络闪断
3 动态的负载均衡过程
    网络闪断/抖动
    broker重启
    订阅方应用重启（消费者）
    客户端扩容
    客户端缩容
幂等性：
同一条消息只有一条。
解决方案：
使用业务id作为消息的key  消费的时候进行判断，未使用过则放过，使用过抛弃

rocketMQ与springboot整合

依赖

<!--RocketMQ相关-->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
</dependency>

配置

rocketmq.name-server=192.168.31.81:9876
rocketmq.producer.group=tanhua

代码实现

//注入rocket
@Autowired
private RocketMQTemplate rocketMQTemplate;
//发送消息
rocketMQTemplate.convertAndSend("tanhua-sso-login", msg);

//获取消息  这里的topic就是上面发送消息的destination
@Service
@RocketMQMessageListener(consumerGroup = "tanhua",topic = "tanhua-sso-login")
public class loginService implements RocketMQListener {

    @Override
    public void onMessage(Object o) {

    }
}

上文的 kafka介绍