consumer

2.进行基本配置pom.xml、application.yml

pom.xml

--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency!--单元测试--dependencygroupIdjunit/groupIdartifactIdjunit/artifactIdversion4.13.2/versionscopetest/scope/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter/artifactId/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdoptionaltrue/optional/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependency/dependenciesapplication.yml

yyyy-MM-dd

消费者启动类ConsumerApplication.class

public

{SpringApplication.run(ConsumerApplication.class,

args);}}生产者启动类PublisherApplication.class

public

{org.springframework.boot.SpringApplication.run(PublisherApplication.class,

args);}

}消费者监听类SpringRabbitListerner.class

Component

listenSimpleQueueMessage(String

msg){System.out.println(简单模式-消费者消费消息msg);}

ExtendWith(SpringExtension.class)

SpringBootTest

world;rabbitTemplate.convertAndSend(simpleQueue,message);}

3.Spring

ExtendWith(SpringExtension.class)

SpringBootTest

world;rabbitTemplate.convertAndSend(simpleQueue,message);}Testpublic

void

world..i;rabbitTemplate.convertAndSend(workQueue,message);}}Testpublic

void

fanout..;rabbitTemplate.convertAndSend(fanoutExchange,,message);}Testpublic

void

direct..;rabbitTemplate.convertAndSend(directExchange,red,message);}}

消费者监听类SpringRabbitListerner.class

Component

listenSimpleQueueMessage(String

msg){System.out.println(简单模式-消费者消费消息msg);}RabbitListener(queues

queue.work)public

msg){System.out.println(工作模式-消费者消费消息msg);}RabbitListener(queues

queue.work)public

msg){System.out.println(工作模式-消费者消费消息2msg);}RabbitListener(queues

queue.fanout1)public

listenFanoutQueueMessage1(String

msg){System.out.println(发布订阅模式-消费者1消费消息msg);}RabbitListener(queues

queue.fanout2)public

listenFanoutQueueMessage2(String

msg){System.out.println(发布订阅模式-消费者2消费消息msg);}RabbitListener(queues

queue.direct1)public

listenDirectQueueMessage(String

msg){System.out.println(路由模式-消费者消费消息msg);}RabbitListener(queues

queue.direct2)public

listenTopicQueueMessage1(String

msg){System.out.println(路由模式-消费者1消费消息msg);}}所用到的配置类

订阅者模式

FanoutExchange(amq.fanout);}Beanpublic

Queue

Queue(queue.fanout1);}Beanpublic

Queue

Queue(queue.fanout2);}Beanpublic

Binding

BindingBuilder.bind(queue1).to(fanoutExchange);}Beanpublic

Binding

BindingBuilder.bind(queue2).to(fanoutExchange);}}路由模式

Configuration

DirectExchange(amq.direct);}Beanpublic

Queue

Queue(queue.direct1);}Beanpublic

Queue

Queue(queue.direct2);}Beanpublic

Binding

BindingBuilder.bind(queue1).to(directExchange).with(yellow);}Beanpublic

Binding

BindingBuilder.bind(queue2).to(directExchange).with(red);}}2025-6-13

2、延迟队列----做定时任务

生产者确认生产者发送消息---到交换机到队列---返回ack给mq生产者发送消息---没到交换机---返回nack给mq生产者发送消息---到交换机没到队列---返回ack触发回调模块剥离代码部分

1.生产者确认机制

HH:mm:ss.SSSlevel:com.example.demo:

debug

同步等待confirm结果直到超时correlated异步回调定义confirmcollback,MQ返回结果时会回调ConfirmCallbackpublisher-returns:

true

功能同样是基于callback机制不过是定义ReturnCallbacktemplate:mandatory:

true

定义消息路由失败时的策略。

true则调用ReturnCallbackfalse则直接丢弃消息

package

org.springframework.amqp.rabbit.connection.CorrelationData;

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;

import

java.util.UUID;RunWith(SpringRunner.class)

SpringBootTest

simpleSendMessage(){//1.准备消息String

msg

amqp;//2.准备CorrelationDate//2.1消息idCorrelationData

correlationData

CorrelationData(UUID.randomUUID().toString());//2.2准备ConfirmCallbackrabbitTemplate.setConfirmCallback((correlationData1,

ack,

cause);}});rabbitTemplate.convertAndSend(amq.direct,yellow,msg,correlationData);}

路由发送失败—触发回调

org.springframework.amqp.rabbit.core.RabbitTemplate;

import

org.springframework.beans.BeansException;

import

org.springframework.context.ApplicationContext;

import

org.springframework.context.ApplicationContextAware;

import

org.springframework.context.annotation.Configuration;Slf4j

Configuration

setApplicationContext(ApplicationContext

applicationContext)

applicationContext.getBean(RabbitTemplate.class);//配置

ReturnCallback

当发送者发送消息到MQMQ返回失败会调用此方法记住是路由发送失败rabbitTemplate.setReturnsCallback(msg

响应码{}

路由key{},msg.getReplyCode(),msg.getReplyText(),msg.getExchange(),msg.getRoutingKey());});}

消费者确认

2.手动处理try-catch然后自己捕捉异常捕捉到了返回给mq服务器nack

解决办法配上retry机制设定本地重试

#关闭ackMQ假定消费者获取消息后会成功处理。

因此消息投递后立即被删除acknowledge-mode:

auto

#自动ack模式由spring检测listener代码是否出现异常没有异常则返回ack有异常则返回nack#acknowledge-mode:

manual

#手动ack,需要在业务代码结束后调用api发送ackretry:enabled:

true

org.springframework.amqp.core.Binding;

import

org.springframework.amqp.core.DirectExchange;

import

org.springframework.amqp.core.Queue;

import

org.springframework.amqp.rabbit.core.RabbitTemplate;

import

org.springframework.amqp.rabbit.retry.MessageRecoverer;

import

org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;

import

org.springframework.context.annotation.Bean;

import

org.springframework.context.annotation.Configuration;Configuration

public

DirectExchange(error.direct);}Beanpublic

Queue

Binding(error.queue,Binding.DestinationType.QUEUE,error.direct,error,null);}Beanpublic

MessageRecoverer

republicMessageRecoverer(RabbitTemplate

rabbitTemplate)

RepublishMessageRecoverer(rabbitTemplate,

error.direct,

初始死信交换机当一个队列中的消息满足下列情况之一时可以成为死信dead

letter

1、消费者使用basic.reject或basic.nack声明消费失败并且消息的requeue的参数设置为false

如果该队列配置了dead-letter-exchange属性指定了一个交换机那么队列中的死信就会投递到这个交换机这个交换机称为死信交换机Dead

Letter

TTL也就是Time-To-Live。

如果一个队列中的消息TTL结束仍未消费则会变成死信ttl超时分为两种情况

模块剥离代码部分

msg){log.info(测试死信队列接收到了延迟消息消息内容为{}msg);}

然后建立延迟交换机和延迟队列的绑定在延迟队列上再绑定死信交换机

附上代码

org.springframework.amqp.core.*;

import

org.springframework.context.annotation.Bean;

import

org.springframework.context.annotation.Configuration;Configuration

public

DirectExchange(ttl.direct);}Beanpublic

Queue

QueueBuilder.durable(ttl.queue).ttl(10000).deadLetterExchange(dl.direct).deadLetterRoutingKey(dl).build();}Beanpublic

Binding

BindingBuilder.bind(ttlQueue).to(ttlExchange).with(ttl);}

}发送消息代码

org.springframework.amqp.core.Message;

import

org.springframework.amqp.core.MessageBuilder;

import

org.springframework.amqp.core.MessageDeliveryMode;

import

org.springframework.amqp.rabbit.connection.CorrelationData;

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;import

java.nio.charset.StandardCharsets;

import

java.util.UUID;RunWith(SpringRunner.class)

SpringBootTest

testTTLMessage(){//1.准备消息Message

message

amqp.getBytes(StandardCharsets.UTF_8))//.setExpiration(5000)

设置延迟时间.setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();//2.发送消息rabbitTemplate.convertAndSend(ttl.direct,ttl,message);}

publisher生产者发送消息

然后再指定时间没有人消息该消息消息自动转为死信并进入指定的延迟队列

1.我发送消息的前提是我mq里面得有已经建好的队列和交换机以及它们之间的绑定关系所以我应该是先启动消费者的启动类将这些交换机和队列存到bean工厂里面去。

形成对象并且在mq里面呈现出来

3.我项目如果是就是一个单一的项目不涉及微服务不分那么多微服务模块这个时候我应该怎么写。

4.3

消息堆积问题1.当生产者发送消息的速度超过了消费者处理消息的速度就会导致队列中的消息堆积直到队列存储消息达到上限。

最早接收到的消息可能成为死信会被丢弃这就是消息堆积问题。

2.解决消息堆积有三种思路1、增加更多消费者提高消费者消费速度就是5种模式的第二种--工作者模式work2、在消费者内开启线程池加快消息处理速度3、扩大队列容积提高堆积上限3.惰性队列特征1、接收到消息后直接存入磁盘而非内存2、消费者要消费消息时才会从磁盘中读取并加载到内存3、支持数百万条的消息存储惰性队列

两种声明方式

org.springframework.amqp.core.Queue;

import

org.springframework.amqp.core.QueueBuilder;

import

org.springframework.context.annotation.Bean;

import

org.springframework.context.annotation.Configuration;Configuration

public

QueueBuilder.durable(lazy.queue).lazy().build();}Beanpublic

Queue

QueueBuilder.durable(normal.queue).build();}}

2、基于注解略

amqp.getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT).build();rabbitTemplate.convertAndSend(lazy.queue,message);}}Testpublic

void

testNormalMessage(){//2.发送消息for

(int

amqp.getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT).build();rabbitTemplate.convertAndSend(normal.queue,message);}}lazy队列的话他是直接发送到磁盘的不进内存

4.4