RocketMQ - 1 消息收发

2024-03-17 15:56:52

RocketMQ - 消息收发

3.0) 消息发送与接收开发流程

谁来发？
发给谁？
怎么发？
发什么？
发的结果是什么？
打扫战场

3.1) 基于Java环境构建消息发送与消息接收基础程序

3.1.1) 单生产者单消费者（OneToOne）

生产者

步骤1：导入坐标

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

步骤2：编写发送消息的程序

package com.itheima.base;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;


/**
 * 生产者，生产消息
 *
 * @author liqp
 * @version 1.0
 */
public class Producer {
    public static void main(String[] args) throws Exception {
        /*
        谁来发？
        发给谁？
        怎么发？
        发什么？
        发的结果是什么？
        打扫战场
         */
        //1.创建一个发送消息的对象
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.设定发送的命名服务器地址
        producer.setNamesrvAddr("192.168.200.169:9876");

        //3.1 启动发送的服务
        producer.start();

        //4. 创建要发送的消息对象,指定topic，指定消息内容
        Message msg = new Message("topic1", "hello rocketmq".getBytes("UTF-8"));

        //3.2 发送消息，获取发送结果
        SendResult result = producer.send(msg);
        System.out.println("返回结果" + result);
        //5. 关闭连接
        producer.shutdown();

    }
}

消费者

package com.itheima.base;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 消费者，消费消息
 *
 * @author liqp
 * @version 1.0
 */
public class Consumer {

    public static void main(String[] args) throws Exception {
        //1.创建一个接收消息的对象Consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.设定接收的命名服务器地址
        consumer.setNamesrvAddr("192.168.200.169:9876");
        //3.设置接收消息对应的topic,对应的sub标签为任意*
        consumer.subscribe("topic1", "*");
        //4.开启监听，用于接收消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt messageExt : list) {
                    System.out.println("消息" + new String(messageExt.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.启动接收消息的服务
        consumer.start();
        System.out.println("接收消息服务已开启运行");
    }
}

3.1.2) 单生产者多消费者（OneToMany）

生产者（发送多条消息）

//1.创建一个发送消息的对象
DefaultMQProducer producer = new DefaultMQProducer("group1");
//2.设定发送的命名服务器地址
producer.setNamesrvAddr("192.168.200.169:9876");

//3.1 启动发送的服务
producer.start();
for (int i = 0; i < 10; i++) {
    //4. 创建要发送的消息对象,指定topic，指定消息内容
    Message msg = new Message("topic1", ("hello rocketmq" + i).getBytes("UTF-8"));

    //3.2 发送消息，获取发送结果
    SendResult result = producer.send(msg);
    System.out.println("返回结果" + result);
}
//5. 关闭连接
producer.shutdown();

消费者(多消费者)

负载均衡模式/集群模式（默认）

consumer.setMessageModel(MessageModel.CLUSTERING);

广播模式

consumer.setMessageModel(MessageModel.BROADCASTING);

3.1.3) 多生产者多消费者（ManyToMany）

多生产者产生的消息可以被同一个消费者消费，也可以被多个消费者消费

3.2) 发送不同类型的消息

3.2.1) 同步消息

即时性较强，重要的消息，且必须有回执的消息，例如短信，通知（转账成功）

SendResult result = producer.send(msg);

3.2.2) 异步消息

即时性较弱，但需要有回执的消息，例如订单中的某些信息

for (int i = 0; i < 10; i++) {
    Message msg = new Message("topic1", ("hello rocketmq" + i).getBytes(StandardCharsets.UTF_8));
    //同步消息
    /*SendResult result = producer.send(msg);
            System.out.println("返回结果" + result);*/

    producer.send(msg, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
            System.out.println(sendResult);
        }

        @Override
        public void onException(Throwable t) {
            System.out.println(t);
        }
    });
}
TimeUnit.SECONDS.sleep(10);

3.2.3) 单向消息

不需要有回执的消息，例如日志类消息

producer.sendOneway(msg);

3.3) 特殊的消息发送

3.3.1) 延时消息

消息发送时并不直接发送到消息服务器，而是根据设定的等待时间到达，起到延时到达的缓冲作用

Message msg = new Message("topic3",("延时消息：hello rocketmq "+i).getBytes("UTF-8"));
//设置当前消息的延时效果
msg.setDelayTimeLevel(3);
SendResult result = producer.send(msg);
System.out.println("返回结果："+result);

目前支持的消息时间
- 秒级：1，5，10，30
- 分级：1~10，20，30
- 时级：1，2
- 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2

3.3.2) 批量消息

发送批量消息

List<Message> msgList = new ArrayList<Message>();
SendResult send = producer.send(msgList);

消息内容总长度不超过4M
消息内容总长度包含如下：
- topic（字符串字节数）
- body （字节数组长度）
- 消息追加的属性（key与value对应字符串字节数）
- 日志（固定20字节）

3.4) 特殊的消息接收

3.4.1) 消息过滤

tag过滤/分类过滤

生产者

Message msg = new Message("topic6","tag2",("消息过滤按照tag：hello rocketmq 2").getBytes("UTF-8"));

消费者

//接收消息的时候，除了制定topic，还可以指定接收的tag,*代表任意tag, 通过 || 分割可以指定多个tag
consumer.subscribe("topic6","tag1 || tag2");

sql过滤/属性过滤/语法过滤

设置broker支持sql过滤
- vim /root/rocket/conf/broker.conf
- 在文件内追加内容enablePropertyFilter=true
- 重启服务
- - cd /root/rocket
  - docker-compose up

生产者

//为消息添加属性
msg.putUserProperty("vip","1");
msg.putUserProperty("age","20");

消费者

//使用消息选择器来过滤对应的属性，语法格式为类SQL语法
consumer.subscribe("topic7", MessageSelector.bySql("age >= 18"));

3.5) 消息发送与接收顺序控制

3.5.1) 错乱的消息顺序

3.5.2）顺序消息

生产者

DefaultMQProducer producer = new DefaultMQProducer("group1");
producer.setNamesrvAddr("192.168.200.169: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) {
            //根据发送的信息不同，选择不同的消息队列
            //根据id来选择一个消息队列的对象，并返回->id得到int值
            int mqIndex = order.getId().hashCode() % list.size();
            return list.get(mqIndex);
        }
    }, null);

    System.out.println(result);
}

producer.shutdown();

消费者

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
consumer.setNamesrvAddr("192.168.200.169: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("接收消息服务已开启运行");

3.6) 事务消息

3.6.1) 正常事务过程与事务补偿过程

3.6.2）事务消息状态

提交状态：允许进入队列，此消息与非事务消息无区别
回滚状态：不允许进入队列，此消息等同于未发送过
中间状态：完成了half消息的发送，未对MQ进行二次状态确认
注意：事务消息仅与生产者有关，与消费者无关

3.6.3）发送事务消息

测试方法

public static void main(String[] args) throws Exception {
    //事务补偿过程
    unknowMsg();

    //事务提交过程
    //commitMsg();

    //事务回滚过程
    //rollbackMsg();

}

事务补偿过程

/**
     * 事务补偿过程
     * @throws Exception
     */
    public static void unknowMsg() throws Exception {
        //事务消息使用的生产者是TransactionMQProducer
        TransactionMQProducer producer = new TransactionMQProducer("group1");
        producer.setNamesrvAddr("192.168.200.169: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("topic12", ("事务消息：hello rocketmq ").getBytes("UTF-8"));
        SendResult result = producer.sendMessageInTransaction(msg, null);
        System.out.println("返回结果：" + result);
        //事务补偿过程必须保障服务器在运行过程中，否则将无法进行正常的事务补偿
//      producer.shutdown();
    }

事务提交过程

/**
     * 事务提交过程
     *
     * @throws Exception
     */
    public static void commitMsg() throws Exception {
        //事务消息使用的生产者是TransactionMQProducer
        TransactionMQProducer producer = new TransactionMQProducer("group1");
        producer.setNamesrvAddr("192.168.200.169:9876");
        //添加本地事务对应的监听
        producer.setTransactionListener(new TransactionListener() {
            //正常事务过程
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                //事务提交状态
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            //事务补偿过程
            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();
    }

事务回滚

/**
     * 事务回滚
     *
     * @throws Exception
     */
    public static void rollbackMsg() throws Exception {
        //事务消息使用的生产者是TransactionMQProducer
        TransactionMQProducer producer = new TransactionMQProducer("group1");
        producer.setNamesrvAddr("192.168.200.169:9876");
        //添加本地事务对应的监听
        producer.setTransactionListener(new TransactionListener() {
            //正常事务过程
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                //事务回滚状态
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }

            //事务补偿过程
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                return null;
            }
        });
        producer.start();

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

码农公寓

RocketMQ - 消息收发

3.0) 消息发送与接收开发流程

3.1) 基于Java环境构建消息发送与消息接收基础程序

3.1.1) 单生产者单消费者（OneToOne）

生产者

消费者

3.1.2) 单生产者多消费者（OneToMany）

生产者（发送多条消息）

消费者(多消费者)

3.1.3) 多生产者多消费者（ManyToMany）

3.2) 发送不同类型的消息

3.2.1) 同步消息

3.2.2) 异步消息

3.2.3) 单向消息

3.3) 特殊的消息发送

3.3.1) 延时消息

3.3.2) 批量消息

3.4) 特殊的消息接收

3.4.1) 消息过滤

tag过滤/分类过滤

生产者

消费者

sql过滤/属性过滤/语法过滤

生产者

消费者

3.5) 消息发送与接收顺序控制

3.5.1) 错乱的消息顺序

3.5.2）顺序消息

生产者

消费者

3.6) 事务消息

3.6.1) 正常事务过程与事务补偿过程

3.6.2）事务消息状态

3.6.3） 发送事务消息

测试方法

事务补偿过程

事务提交过程

事务回滚

相关文章

3.6.3）发送事务消息