RocketMQ发送消息和消费消息测试类

====================二期实现
导入依赖

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

测试代码

package com.example.springbootproject;

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.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
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.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.List;

@SpringBootTest
@RunWith(SpringRunner.class)
@Slf4j
public class mqtest {
/**
* 生产者
* @throws Exception
*/
@Test
public void test() throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("GROUP_DEMO");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.setSendLatencyFaultEnable(true);

producer.start();
for (int i = 0; i < 10; i++) {
Message msg = new Message("TOPIC_TEST",
"TAGE_TEST",
("ROCKETMQ000" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
//发送核心方法
//同步
SendResult send = producer.send(msg);
//异步
producer.send(msg, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.println("===" + sendResult);
}

@Override
public void onException(Throwable throwable) {
System.out.println("e:" + throwable);
}
});
//单向
producer.sendOneway(msg);

System.out.println("send:%s%n" + send);
}
//producer.shutdown();
}

/**
* 消费者
* @throws Exception
*/
@Test
public void consume() throws Exception {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group_consumer");
consumer.setNamesrvAddr("127.0.0.1:9876");
//订阅Topic去消费生产者产生的消息。
consumer.subscribe("TOPIC_TEST", "*");//tag tagA|tagB|tagC
consumer.setMessageModel(MessageModel.CLUSTERING);

consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
try {
for (MessageExt item : list) {
String topic = item.getTopic();
String tags = item.getTags();
String msgBody = new String(item.getBody(), "utf-8");
System.out.println("收到消息:topic:" + topic + ",tags:" + tags + ",msg:" + msgBody);

}
} catch (Exception e) {
log.error("e:", e);
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//启动消费者
consumer.start();
System.out.println("consumer start");
}


}

====================一期实现

public class RockMQSendMessageTest {

//发送消息
public static void main(String[] args) throws Exception {
//1创建消息生产者,并且设置生成组名
DefaultMQProducer producer = new DefaultMQProducer("myproducer-group");
//2为生产者设置NameServer的地址 mq ip:port port可以在本地跑起来mq
producer.setNamesrvAddr(".......");
//3 启动生产者
producer.start();
//4 构建消息对象,主要设置消息的主题 标签 内容 topic tags 内容
Message message = new Message("myTopic", "myTag", ("Test jfdlfd").getBytes());
//5发送消息
SendResult result = producer.send(message, 1000);
System.out.println(result);
//6 关闭生产者
producer.shutdown();
}
}
public class RocketMQReceviceMessageTest {
//接收消息
public static void main(String[] args) throws Exception {
//1创建消费者,并且为其制定消费者组名
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myconsumer-group");
//2为消费者设置NameServer的地址ip:port可以在本地跑起来mq
consumer.setNamesrvAddr(".......");
//3制定消费者订阅的主题和标签
consumer.subscribe("myTopic", "*");
//4设置一个回调函数,并在函数编写接收到消息之后的处理方法
consumer.registerMessageListener(new MessageListenerConcurrently() {
//j处理获取到的消息
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
//消费逻辑
System.out.println("Message====>" + list);
//消息成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//5启动消费者
consumer.start();
System.out.println("启动消费成功");

}
}