SpringCloudAlibaba:6.2RocketMQ的普通消息的使用

简介

普通消息也叫并发消息，是发送效率最高，使用最多的一种

依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><parent><artifactId>java_sc_alibaba</artifactId><groupId>jkw.life</groupId><version>1.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>test-rocketmq8009</artifactId><properties><maven.compiler.source>17</maven.compiler.source><maven.compiler.target>17</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><dependencies><!-- rocketmq--><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>5.1.0</version></dependency><!-- SpringMVC--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- test--><dependency><groupId>junit</groupId><artifactId>junit</artifactId><scope>test</scope></dependency></dependencies></project>

创建topic

mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t testtopic

测试类

package send_message;import lombok.extern.slf4j.Slf4j;
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.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.junit.Test;
import java.nio.charset.StandardCharsets;
import java.util.List;
@Slf4j
public class Test_01 {/*** 消息生成者【普通消息】* 创建topic：* mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t testtopic*/@Testpublic void producer() throws MQClientException, MQBrokerException, RemotingException, InterruptedException {//1.初始化生产者【生产者组名】DefaultMQProducer producer = new DefaultMQProducer("SyncProducer");//2.rocketmq地址producer.setNamesrvAddr("192.168.66.101:9876");//3.启动生产者producer.start();for (int i = 0; i < 10; i++) {//4.初始化消息对象【topic主题/标记过滤/消息体】Message message = new Message("testtopic", "Tags", (i + "_syncProducer").getBytes(StandardCharsets.UTF_8));//5.生产者发送消息SendResult send = producer.send(message);System.out.println(i + "消息发送成功:" + send);}//6.关闭生产者producer.shutdown();}/*** 消费者*/@Testpublic void consumer() throws MQClientException, InterruptedException {//1.初始化消费者【消费者组名】DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("SyncProducer");//2.rocketmq地址consumer.setNamesrvAddr("192.168.66.101:9876");//3.订阅主题consumer.subscribe("testtopic", "*");//4.监听消息consumer.setMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {for (MessageExt msg : list) {System.out.println("消费成功" + msg);}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//消费成功}});//5.启动消费者consumer.start();//6.永远运行下去Thread.sleep(Long.MAX_VALUE);}
}