章目录

方法上加上RabbitListenner表明需要监听的queue)

常见参数汇总

再rabbitMQ中有两个重要的组件。

exchange交换机queue队列。

交换机用于路由消息简单来说就是接收客户端传递的消息转发到queue中。

队列做的事情就是存储消息。

当遇到死掉的消息时我们通常会将这些死信转发到新的交换机中这个交换机就叫做死信交换机而配合死信交换机存储信息的队列叫做死信队列

1.2

死信队列在构建延迟队列时有巨大作用。

比如用户购票订单30min不支付就过期。

在rabbitMQ中可以这样实现

如果超过30min消息未被消费消息过期成为死信存储死信队列通知服务取消订单

如何编码

整个流程中出现了三方。

在编写代码时我们可以分三个大类分别是producer,consumer1,consumer2。

其消息传递顺序如下

producer

我们可以在编写consumer1的时候完成exchange和queue的创建与绑定

2.2

dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.9.0/version/dependency2.3

工具类编写

创建Connection工厂配置工厂创建链接获取channel通过channel创建exchange,

queue,

因此我们可以先创建工具类帮我们获取channel以此减少开发代码量

public

设置工厂factory.setHost(your_ip);factory.setUsername(your_username);factory.setPassword(your_password);factory.setVirtualHost(/);

connection

factory.newConnection();Channel

channel

connection.createChannel();return

channel;}

普通交换机channel.exchangeDeclare(EXCHANGE_NAME,

false,

配置normal_queue连接的dead_exchangemap.put(x-dead-letter-exchange,

设置normal_queue的消息过期时间map.put(x-message-ttl,

10000);//

lisimap.put(x-dead-letter-routing-key,

lisi);//

设置队列最大长度channel.queueDeclare(QUEUE_NAME,

false,

绑定普通交换机和普通队列channel.queueBind(QUEUE_NAME,

EXCHANGE_NAME,

死信交换机channel.exchangeDeclare(DEAD_EXCHANGE_NAME,

false,

死信队列channel.queueDeclare(DEAD_QUEUE_NAME,

false,

绑定死信交换机和死信队列channel.queueBind(DEAD_QUEUE_NAME,

DEAD_EXCHANGE_NAME,

监听channel.basicConsume(QUEUE_NAME,

false,

监听channel.basicConsume(DEAD_QUEUE_NAME,

true,

;channel.basicPublish(EXCHANGE_NAME,

zhangsan,null,msg.getBytes());}}

}3.整合springboot

dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactIdversion2.3.9.RELEASE/version/dependency3.3

package

org.springframework.amqp.core.*;

import

org.springframework.beans.factory.annotation.Qualifier;

import

org.springframework.context.annotation.Bean;

import

org.springframework.context.annotation.Configuration;Configuration

public

ExchangeBuilder.directExchange(EXCHANGE_NAME).autoDelete().build();

return

DirectExchange(EXCHANGE_NAME);}//

注册死信交换机Bean(deadExchange)public

DirectExchange

ExchangeBuilder.directExchange(DEAD_EXCHANGE_NAME).autoDelete().build();}//

Queue

QueueBuilder.nonDurable(QUEUE_NAME1).withArgument(x-message-ttl,

10000).withArgument(x-dead-letter-exchange,

DEAD_EXCHANGE_NAME).withArgument(x-dead-letter-routing-key,

YD).build();}//

QueueBuilder.nonDurable(QUEUE_NAME2).withArgument(x-message-ttl,

40000).withArgument(x-dead-letter-exchange,

DEAD_EXCHANGE_NAME).withArgument(x-dead-letter-routing-key,

YD).build();}//

QueueBuilder.nonDurable(DEAD_QUEUE_NAME).build();}//

队列Beanpublic

queueABindingX(Qualifier(queueA)

Queue

BindingBuilder.bind(queueA).to(xExchange).with(XA).and(null);}Beanpublic

Binding

queueBBindingX(Qualifier(queueB)

Queue

BindingBuilder.bind(queueB).to(xExchange).with(XB).and(null);}//

绑定死信交换机,

queueDBindingY(Qualifier(queueD)

Queue

BindingBuilder.bind(queueD).to(deadExchange).with(YD).and(null);}

3.4

消息通过XA路由到队列rabbitTemplate.convertAndSend(X,

XA,

方法上加上RabbitListenner表明需要监听的queue)

Component

message.getBody();System.out.println(new

String(body));}

live时间过期时间x-dead-letter-routing-key死信交换机路由keyx-max-length队列最大长度