С сегодняшнего дня закатывайте рукава и следуйте за мной, если у вас есть какие-либо вопросы, пожалуйста, свяжитесь со мной в личном сообщении
предисловие
Использование Spring Boot может быть очень удобным и быстрым для сборки проекта, так что нам не нужно заботиться о совместимости между фреймворками, применимыми версиями и другими вопросами, мы хотим использовать что угодно, просто добавляем конфигурацию.
Совет: Ниже приведен основной текст этой статьи, следующие случаи приведены для справки.
1. Техническое введение
1. Что такое очередь сообщений?
Промежуточное программное обеспечение очереди сообщений — важный компонент распределенных систем, который в основном решает такие проблемы, как связывание приложений, асинхронные сообщения и сокращение трафика. Внедрите высокопроизводительную, доступную, масштабируемую и, в конечном счете, согласованную архитектуру.
2. Что такое RocketMq?
RocketMQ — это промежуточное программное обеспечение для сообщений модели очереди, которое обладает характеристиками высокой производительности, высокой надежности, высокой производительности в реальном времени и распределенности. Это распределенная система обмена сообщениями, разработанная на языке Java командой Alibaba.
2. Используйте шаги
1. Познакомьтесь с библиотекой maven
Код выглядит следующим образом (пример):
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.4.1</version>
<relativePath/>
</parent>
<dependencies>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.1.1</version>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.8.0</version>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-common</artifactId>
<version>4.8.0</version>
</dependency>
</dependencies>
2. Инкапсулировать класс инструмента RocketMq
Код выглядит следующим образом (пример):
package com.hyh.rocketmq.helper;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
/**
* RocketMq助手
*
* @Author: heyuhua
* @Date: 2020/1/8 10:03
*/
@Component
public class RocketMqHelper {
/**
* 日志
*/
private static final Logger LOG = LoggerFactory.getLogger(RocketMqHelper.class);
/**
* rocketmq模板注入
*/
@Autowired
private RocketMQTemplate rocketMQTemplate;
@PostConstruct
public void init() {
LOG.info("---RocketMq助手初始化---");
}
/**
* 发送异步消息
*
* @param topic 消息Topic
* @param message 消息实体
*/
public void asyncSend(Enum topic, Message<?> message) {
asyncSend(topic.name(), message, getDefaultSendCallBack());
}
/**
* 发送异步消息
*
* @param topic 消息Topic
* @param message 消息实体
* @param sendCallback 回调函数
*/
public void asyncSend(Enum topic, Message<?> message, SendCallback sendCallback) {
asyncSend(topic.name(), message, sendCallback);
}
/**
* 发送异步消息
*
* @param topic 消息Topic
* @param message 消息实体
*/
public void asyncSend(String topic, Message<?> message) {
rocketMQTemplate.asyncSend(topic, message, getDefaultSendCallBack());
}
/**
* 发送异步消息
*
* @param topic 消息Topic
* @param message 消息实体
* @param sendCallback 回调函数
*/
public void asyncSend(String topic, Message<?> message, SendCallback sendCallback) {
rocketMQTemplate.asyncSend(topic, message, sendCallback);
}
/**
* 发送异步消息
*
* @param topic 消息Topic
* @param message 消息实体
* @param sendCallback 回调函数
* @param timeout 超时时间
*/
public void asyncSend(String topic, Message<?> message, SendCallback sendCallback, long timeout) {
rocketMQTemplate.asyncSend(topic, message, sendCallback, timeout);
}
/**
* 发送异步消息
*
* @param topic 消息Topic
* @param message 消息实体
* @param sendCallback 回调函数
* @param timeout 超时时间
* @param delayLevel 延迟消息的级别
*/
public void asyncSend(String topic, Message<?> message, SendCallback sendCallback, long timeout, int delayLevel) {
rocketMQTemplate.asyncSend(topic, message, sendCallback, timeout, delayLevel);
}
/**
* 发送顺序消息
*
* @param message
* @param topic
* @param hashKey
*/
public void syncSendOrderly(Enum topic, Message<?> message, String hashKey) {
syncSendOrderly(topic.name(), message, hashKey);
}
/**
* 发送顺序消息
*
* @param message
* @param topic
* @param hashKey
*/
public void syncSendOrderly(String topic, Message<?> message, String hashKey) {
LOG.info("发送顺序消息，topic:" + topic + ",hashKey:" + hashKey);
rocketMQTemplate.syncSendOrderly(topic, message, hashKey);
}
/**
* 发送顺序消息
*
* @param message
* @param topic
* @param hashKey
* @param timeout
*/
public void syncSendOrderly(String topic, Message<?> message, String hashKey, long timeout) {
LOG.info("发送顺序消息，topic:" + topic + ",hashKey:" + hashKey + ",timeout:" + timeout);
rocketMQTemplate.syncSendOrderly(topic, message, hashKey, timeout);
}
/**
* 默认CallBack函数
*
* @return
*/
private SendCallback getDefaultSendCallBack() {
return new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
LOG.info("---发送MQ成功---");
}
@Override
public void onException(Throwable throwable) {
LOG.error("---发送MQ失败---"+throwable.getMessage(), throwable.getMessage());
}
};
}
@PreDestroy
public void destroy() {
LOG.info("---RocketMq助手注销---");
}
}
3. Файл конфигурации
Код выглядит следующим образом (пример):
server:
port: 8088
#rocketmq配置
rocketmq:
name-server: 127.0.0.1:9876
# 生产者配置
producer:
isOnOff: on
# 发送同一类消息的设置为同一个group，保证唯一
group: hyh-rocketmq-group
groupName: hyh-rocketmq-group
# 服务地址
namesrvAddr: 127.0.0.1:9876
# 消息最大长度 默认1024*4(4M)
maxMessageSize: 4096
# 发送消息超时时间,默认3000
sendMsgTimeout: 3000
# 发送消息失败重试次数，默认2
retryTimesWhenSendFailed: 2
4. Модульное тестирование
Код теста выглядит следующим образом (пример):
@Autowired
private RocketMqHelper rocketMqHelper;
@Test
public void testProducter() {
Person person = new Person();
person.setName("heyuhua");
person.setAge(25);
rocketMqHelper.asyncSend("PERSON_ADD", MessageBuilder.withPayload(person).build());
}
Код мониторинга сообщений выглядит следующим образом (пример):
package com.hyh.core.listener;
import com.hyh.core.po.Person;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
/**
* @Author: heyuhua
* @Date: 2021/1/8 15:55
*/
@Component
@RocketMQMessageListener(consumerGroup = "${rocketmq.producer.groupName}", topic = "PERSON_ADD")
public class PersonMqListener implements RocketMQListener<Person> {
@Override
public void onMessage(Person person) {
System.out.println("接收到消息，开始消费..name:" + person.getName() + ",age:" + person.getAge());
}
}
Суммировать
Это кажется простым? Следуйте за мной, чтобы показать вам более продвинутое использование RocketMq Адрес источника:Нажмите здесь, чтобы просмотреть исходный код.