-
Spring Boot整合RabbitMQ
写在开头
最近在搭一套基于SpringBoot的项目,用到了ssm+mysql+rabbitmq+redis。除了rabbitmq之外,其他几个都很快整合好了,唯独rabbitmq找了不少资料,才最终整合好,达到了预期。特此将过程记录下来,供参考。
整合流程
整合流程中的代码都为整合的关键配置及其使用。至于SpringBoot的基本配置,请参考Spring Boot Quick Start。
配置文件
- pom.xml
<!-- rabbit-mq -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
- application.yml
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
virtual-host: /
listener:
simple:
acknowledge-mode: manual # 手动应答
concurrency: 5 # 消费端最小并发数
max-concurrency: 10 # 消费端最大并发数
prefetch: 5 # 一次请求中预处理的消息数量
cache:
channel:
size: 50 # 缓存的channel数量
### 自定义配置
mq:
defaultExchange: amqpExchange # 默认交换器
queue: queue # 队列名
routeKey: queue_key # 路由key
- MQProperties
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
@Component
@ConfigurationProperties(prefix = "mq")
public class MQProperties {
private String defaultExchange;
private String routeKey;
private String queue;
public String getDefaultExchange() {
return defaultExchange;
}
public void setDefaultExchange(String defaultExchange) {
this.defaultExchange = defaultExchange;
}
public String getRouteKey() {
return routeKey;
}
public void setRouteKey(String routeKey) {
this.routeKey = routeKey;
}
public String getQueue() {
return queue;
}
public void setQueue(String queue) {
this.queue = queue;
}
}
RabbitMQ配置
import com.switchvov.rabbitmq.constant.MQProperties;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
@EnableRabbit
public class RabbitMQConfig {
@Autowired
private MQProperties mqProperties;
@Bean
public Queue queue() {
boolean durable = true;
boolean exclusive = false;
boolean autoDelete = false;
return new Queue(mqProperties.getQueue(), durable, exclusive, autoDelete);
}
@Bean
public DirectExchange defaultExchange() {
boolean durable = true;
boolean autoDelete = false;
return new DirectExchange(mqProperties.getDefaultExchange(), durable, autoDelete);
}
@Bean
public Binding binding() {
return BindingBuilder.bind(queue())
.to(defaultExchange())
.with(mqProperties.getRouteKey());
}
}
RabbitMQ生产者
import com.switchvov.rabbitmq.constant.MQProperties;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMQTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private MQProperties mqProperties;
@Test
public void testSendMessage() {
rabbitTemplate.convertAndSend(mqProperties.getDefaultExchange(),
mqProperties.getRouteKey(), "发送了一条信息");
}
}
RabbitMQ消费者
import com.switchvov.rabbitmq.common.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
@Service
public class RabbitMQService {
private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMQService.class);
@RabbitListener(queues = "${mq.queue}")
public void receive(String payload, Channel channel,
@Header(AmqpHeaders.DELIVERY_TAG) long tag) {
LOGGER.info("消费内容为:{}", payload);
RabbitMQUtils.askMessage(channel, tag, LOGGER);
}
}
手动应答简单工具类
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import java.io.IOException;
public final class RabbitMQUtils {
public static void askMessage(Channel channel, long tag, final Logger logger) {
askMessage(channel, tag, logger, false);
}
public static void askMessage(Channel channel, long tag, final Logger logger, boolean multiple) {
try {
channel.basicAck(tag, multiple);
} catch (IOException e) {
logger.error("RabbitMQ,IO异常,异常原因为:{}", e.getMessage());
}
}
public static void rejectMessage(Channel channel, long tag, final Logger logger) {
rejectMessage(channel, tag, logger, false, false);
}
public static void rejectAndBackMQ(Channel channel, long tag, final Logger logger) {
rejectMessage(channel, tag, logger, false, true);
}
public static void rejectMessage(Channel channel, long tag, final Logger logger, boolean multiple, boolean request) {
try {
channel.basicNack(tag, multiple, request);
} catch (IOException e) {
logger.error("RabbitMQ,IO异常,异常原因为:{}", e.getMessage());
}
}
}
出处:https://www.cnblogs.com/switchvov/p/15070110.html
最新更新
python爬虫及其可视化
使用python爬取豆瓣电影短评评论内容
nodejs爬虫
Python正则表达式完全指南
爬取豆瓣Top250图书数据
shp 地图文件批量添加字段
爬虫小试牛刀(爬取学校通知公告)
【python基础】函数-初识函数
【python基础】函数-返回值
HTTP请求:requests模块基础使用必知必会
SQL SERVER中递归
2个场景实例讲解GaussDB(DWS)基表统计信息估
常用的 SQL Server 关键字及其含义
动手分析SQL Server中的事务中使用的锁
openGauss内核分析:SQL by pass & 经典执行
一招教你如何高效批量导入与更新数据
天天写SQL,这些神奇的特性你知道吗?
openGauss内核分析:执行计划生成
[IM002]Navicat ODBC驱动器管理器 未发现数据
初入Sql Server 之 存储过程的简单使用
uniapp/H5 获取手机桌面壁纸 (静态壁纸)
[前端] DNS解析与优化
为什么在js中需要添加addEventListener()?
JS模块化系统
js通过Object.defineProperty() 定义和控制对象
这是目前我见过最好的跨域解决方案!
减少回流与重绘
减少回流与重绘
如何使用KrpanoToolJS在浏览器切图
performance.now() 与 Date.now() 对比