定义监听器消费消息
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "ly.search.insert.queue", durable = "true"),
exchange = @Exchange(name = "ly.item.exchange",
type = ExchangeTypes.TOPIC,
ignoreDeclarationExceptions = "true"),
key = {"item.insert", "item.update"}
))
public void listenInsert(Long id) {
//监听新增或更新
if (id != null) {
searchService.insertOrUpdate(id);
}
}
自动创建,queue 和 exchange 绑定
@RabbitListener注解指定目标方法来作为消费消息的方法,通过注解参数指定所监听的队列或者Binding。
@QueueBinding 将交换机和队列绑定
key = {"item.insert", "item.update"} 返回绑定的路由密钥或模式,多个元素将导致多个绑定
@Queue声明队列 (durable = "true" 表示持久化的)
@Exchange声明交换机(type = ExchangeTypes.TOPIC 表示交换机类型)
配置文件
spring:
rabbitmq:
host: 192.168.1.188
username: sunfeng
password: sunfeng
virtual-host: /sunfeng
template:
retry:
enabled: true
initial-interval: 10000ms
max-interval: 30000ms
multiplier: 2
exchange: ly.item.exchange
publisher-confirms: true
pom文件
<!--amqp依赖-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
消息发送
@Autowired
private AmqpTemplate amqpTemplate; 注入模板
/**
* 封装发送到消息队列的方法
*
* @param id
* @param type 发送消息的类型
*/
private void sendMessage(Long id, String type) {
log.info("发送消息到mq");
try {
amqpTemplate.convertAndSend("item." + type, id);
} catch (Exception e) {
log.error("{}商品消息发送异常,商品ID:{}", type, id, e);
}
}
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。