您的位置：首页 > 编程语言 > Java开发

springboot+rabbitmq整合示例程

2017-05-11 17:42 696 查看
关于什么是rabbitmq，请看另一篇文：
http://www.cnblogs.com/boshen-hzb/p/6840064.html
一、新建maven工程：springboot-rabbitmq



二、引入springboot和rabbitmq的依赖

<projectxmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion>
<groupId>com.springboot.rabbitmq</groupId>
<artifactId>springboot-rabbitmq</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>springboot-rabbitmq</name>
<description>springboot-rabbitmq</description>

<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.4.1.RELEASE</version>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
</dependencies>
</project>


spring-boot-starter-test是为了后面写测试类用，

spring-boot-starter-amqp才是真正的使用rabbitmq的依赖

三、在src/main/resources里面新增application.properties
该配置文件主要是对rabbimq的配置信息



spring.application.name=springboot-rabbitmq
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.virtual-host=/



四、新建springboot主类Application

该类初始化创建队列、转发器，并把队列绑定到转发器

packagecom.rabbit;

importorg.springframework.amqp.core.Binding;
importorg.springframework.amqp.core.BindingBuilder;
importorg.springframework.amqp.core.FanoutExchange;
importorg.springframework.amqp.core.Queue;
importorg.springframework.amqp.core.TopicExchange;
importorg.springframework.boot.SpringApplication;
importorg.springframework.boot.autoconfigure.SpringBootApplication;
importorg.springframework.context.annotation.Bean;

@SpringBootApplication
publicclassApplication{
finalstaticStringqueueName="hello";

@Bean
publicQueuehelloQueue(){
returnnewQueue("hello");
}

@Bean
publicQueueuserQueue(){
returnnewQueue("user");
}

//===============以下是验证topicExchange的队列==========
@Bean
publicQueuequeueMessage(){
returnnewQueue("topic.message");
}

@Bean
publicQueuequeueMessages(){
returnnewQueue("topic.messages");
}
//===============以上是验证topicExchange的队列==========

//===============以下是验证FanoutExchange的队列==========
@Bean
publicQueueAMessage(){
returnnewQueue("fanout.A");
}

@Bean
publicQueueBMessage(){
returnnewQueue("fanout.B");
}

@Bean
publicQueueCMessage(){
returnnewQueue("fanout.C");
}
//===============以上是验证FanoutExchange的队列==========

@Bean
TopicExchangeexchange(){
returnnewTopicExchange("exchange");
}
@Bean
FanoutExchangefanoutExchange(){
returnnewFanoutExchange("fanoutExchange");
}

/**
*将队列topic.message与exchange绑定，binding_key为topic.message,就是完全匹配
*@paramqueueMessage
*@paramexchange
*@return
*/
@Bean
BindingbindingExchangeMessage(QueuequeueMessage,TopicExchangeexchange){
returnBindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
}

/**
*将队列topic.messages与exchange绑定，binding_key为topic.#,模糊匹配
*@paramqueueMessage
*@paramexchange
*@return
*/
@Bean
BindingbindingExchangeMessages(QueuequeueMessages,TopicExchangeexchange){
returnBindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
}

@Bean
BindingbindingExchangeA(QueueAMessage,FanoutExchangefanoutExchange){
returnBindingBuilder.bind(AMessage).to(fanoutExchange);
}

@Bean
BindingbindingExchangeB(QueueBMessage,FanoutExchangefanoutExchange){
returnBindingBuilder.bind(BMessage).to(fanoutExchange);
}

@Bean
BindingbindingExchangeC(QueueCMessage,FanoutExchangefanoutExchange){
returnBindingBuilder.bind(CMessage).to(fanoutExchange);
}

publicstaticvoidmain(String[]args)throwsException{
SpringApplication.run(Application.class,args);
}
}


五、各种情景实现

1、最简单的hello生产和消费实现（单生产者和单消费者）

生产者：

packagecom.rabbit.hello;

importjava.util.Date;

importorg.springframework.amqp.core.AmqpTemplate;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.stereotype.Component;

@Component
publicclassHelloSender1{

@Autowired
privateAmqpTemplaterabbitTemplate;

publicvoidsend(){
StringsendMsg="hello1"+newDate();
System.out.println("Sender1:"+sendMsg);
this.rabbitTemplate.convertAndSend("helloQueue",sendMsg);
}

}


消费者:

packagecom.rabbit.hello;

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;
importorg.springframework.amqp.rabbit.annotation.RabbitListener;
importorg.springframework.stereotype.Component;

@Component
@RabbitListener(queues="helloQueue")
publicclassHelloReceiver1{

@RabbitHandler
publicvoidprocess(Stringhello){
System.out.println("Receiver1:"+hello);
}

}


controller：

packagecom.rabbit.controller;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.web.bind.annotation.PostMapping;
importorg.springframework.web.bind.annotation.RequestBody;
importorg.springframework.web.bind.annotation.RequestMapping;
importorg.springframework.web.bind.annotation.RestController;

importcom.rabbit.hello.HelloSender1;

@RestController
@RequestMapping("/rabbit")
publicclassRabbitTest{

@Autowired
privateHelloSender1helloSender1;
@Autowired
privateHelloSender1helloSender2;

@PostMapping("/hello")
publicvoidhello(){
helloSender1.send();
}
}


启动程序，执行：



结果如下：

Sender1:hello1ThuMay1117:23:31CST2017
Receiver2:hello1ThuMay1117:23:31CST2017


2、单生产者-多消费者

生产者：

packagecom.rabbit.hello;

importjava.util.Date;

importorg.springframework.amqp.core.AmqpTemplate;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.stereotype.Component;

@Component
publicclassHelloSender1{

@Autowired
privateAmqpTemplaterabbitTemplate;

publicvoidsend(Stringmsg){
StringsendMsg=msg+newDate();
System.out.println("Sender1:"+sendMsg);
this.rabbitTemplate.convertAndSend("helloQueue",sendMsg);
}

}


消费者1：

packagecom.rabbit.hello;

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;
importorg.springframework.amqp.rabbit.annotation.RabbitListener;
importorg.springframework.stereotype.Component;

@Component
@RabbitListener(queues="helloQueue")
publicclassHelloReceiver1{

@RabbitHandler
publicvoidprocess(Stringhello){
System.out.println("Receiver1:"+hello);
}

}


消费者2：

packagecom.rabbit.hello;

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;
importorg.springframework.amqp.rabbit.annotation.RabbitListener;
importorg.springframework.stereotype.Component;

@Component
@RabbitListener(queues="helloQueue")
publicclassHelloReceiver2{

@RabbitHandler
publicvoidprocess(Stringhello){
System.out.println("Receiver2:"+hello);
}

}


controller:

packagecom.rabbit.controller;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.web.bind.annotation.PostMapping;
importorg.springframework.web.bind.annotation.RequestBody;
importorg.springframework.web.bind.annotation.RequestMapping;
importorg.springframework.web.bind.annotation.RestController;

importcom.rabbit.hello.HelloSender1;

@RestController
@RequestMapping("/rabbit")
publicclassRabbitTest{

@Autowired
privateHelloSender1helloSender1;
@Autowired
privateHelloSender1helloSender2;

@PostMapping("/hello")
publicvoidhello(){
helloSender1.send("hello1");
}

/**
*单生产者-多消费者
*/
@PostMapping("/oneToMany")
publicvoidoneToMany(){
for(inti=0;i<10;i++){
helloSender1.send("hellomsg:"+i);
}

}
}


用post方式执行：
http://127.0.0.1:8080/rabbit/oneToMany
结果如下：

Sender1:hellomsg:0ThuMay1117:37:59CST2017
Sender1:hellomsg:1ThuMay1117:37:59CST2017
Sender1:hellomsg:2ThuMay1117:37:59CST2017
Sender1:hellomsg:3ThuMay1117:37:59CST2017
Sender1:hellomsg:4ThuMay1117:37:59CST2017
Sender1:hellomsg:5ThuMay1117:37:59CST2017
Sender1:hellomsg:6ThuMay1117:37:59CST2017
Sender1:hellomsg:7ThuMay1117:37:59CST2017
Sender1:hellomsg:8ThuMay1117:37:59CST2017
Sender1:hellomsg:9ThuMay1117:37:59CST2017
Receiver2:hellomsg:1ThuMay1117:37:59CST2017
Receiver1:hellomsg:0ThuMay1117:37:59CST2017
Receiver1:hellomsg:3ThuMay1117:37:59CST2017
Receiver1:hellomsg:4ThuMay1117:37:59CST2017
Receiver1:hellomsg:5ThuMay1117:37:59CST2017
Receiver2:hellomsg:2ThuMay1117:37:59CST2017
Receiver1:hellomsg:6ThuMay1117:37:59CST2017
Receiver2:hellomsg:7ThuMay1117:37:59CST2017
Receiver2:hellomsg:8ThuMay1117:37:59CST2017
Receiver1:hellomsg:9ThuMay1117:37:59CST2017


从以上结果可知，生产者发送的10条消息，分别被两个消费者接收了

3、多生产者-多消费者

生产者1：

packagecom.rabbit.hello;

importjava.util.Date;

importorg.springframework.amqp.core.AmqpTemplate;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.stereotype.Component;

@Component
publicclassHelloSender1{

@Autowired
privateAmqpTemplaterabbitTemplate;

publicvoidsend(Stringmsg){
StringsendMsg=msg+newDate();
System.out.println("Sender1:"+sendMsg);
this.rabbitTemplate.convertAndSend("helloQueue",sendMsg);
}

}


生产者2：

packagecom.rabbit.hello;

importjava.util.Date;

importorg.springframework.amqp.core.AmqpTemplate;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.stereotype.Component;

@Component
publicclassHelloSender2{

@Autowired
privateAmqpTemplaterabbitTemplate;

publicvoidsend(Stringmsg){
StringsendMsg=msg+newDate();
System.out.println("Sender2:"+sendMsg);
this.rabbitTemplate.convertAndSend("helloQueue",sendMsg);
}

}


消费者1：

packagecom.rabbit.hello;

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;
importorg.springframework.amqp.rabbit.annotation.RabbitListener;
importorg.springframework.stereotype.Component;

@Component
@RabbitListener(queues="helloQueue")
publicclassHelloReceiver1{

@RabbitHandler
publicvoidprocess(Stringhello){
System.out.println("Receiver1:"+hello);
}

}


消费者2：

packagecom.rabbit.hello;

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;
importorg.springframework.amqp.rabbit.annotation.RabbitListener;
importorg.springframework.stereotype.Component;

@Component
@RabbitListener(queues="helloQueue")
publicclassHelloReceiver2{

@RabbitHandler
publicvoidprocess(Stringhello){
System.out.println("Receiver2:"+hello);
}

}


controller:

/**
*多生产者-多消费者
*/
@PostMapping("/manyToMany")
publicvoidmanyToMany(){
for(inti=0;i<10;i++){
helloSender1.send("hellomsg:"+i);
helloSender2.send("hellomsg:"+i);
}

}


用post方式执行：
http://127.0.0.1:8080/rabbit/manyToMany
结果如下：

Sender1:hellomsg:0FriMay1209:08:50CST2017
Sender2:hellomsg:0FriMay1209:08:50CST2017
Sender1:hellomsg:1FriMay1209:08:50CST2017
Sender2:hellomsg:1FriMay1209:08:50CST2017
Sender1:hellomsg:2FriMay1209:08:50CST2017
Sender2:hellomsg:2FriMay1209:08:50CST2017
Sender1:hellomsg:3FriMay1209:08:50CST2017
Sender2:hellomsg:3FriMay1209:08:50CST2017
Sender1:hellomsg:4FriMay1209:08:50CST2017
Sender2:hellomsg:4FriMay1209:08:50CST2017
Sender1:hellomsg:5FriMay1209:08:50CST2017
Sender2:hellomsg:5FriMay1209:08:50CST2017
Sender1:hellomsg:6FriMay1209:08:50CST2017
Sender2:hellomsg:6FriMay1209:08:50CST2017
Sender1:hellomsg:7FriMay1209:08:50CST2017
Sender2:hellomsg:7FriMay1209:08:50CST2017
Sender1:hellomsg:8FriMay1209:08:50CST2017
Sender2:hellomsg:8FriMay1209:08:50CST2017
Sender1:hellomsg:9FriMay1209:08:50CST2017
Sender2:hellomsg:9FriMay1209:08:50CST2017
Receiver2:hellomsg:0FriMay1209:08:50CST2017
Receiver1:hellomsg:0FriMay1209:08:50CST2017
Receiver2:hellomsg:1FriMay1209:08:50CST2017
Receiver1:hellomsg:1FriMay1209:08:50CST2017
Receiver2:hellomsg:2FriMay1209:08:50CST2017
Receiver1:hellomsg:2FriMay1209:08:50CST2017
Receiver2:hellomsg:3FriMay1209:08:50CST2017
Receiver1:hellomsg:3FriMay1209:08:50CST2017
Receiver2:hellomsg:4FriMay1209:08:50CST2017
Receiver1:hellomsg:4FriMay1209:08:50CST2017
Receiver2:hellomsg:5FriMay1209:08:50CST2017
Receiver1:hellomsg:5FriMay1209:08:50CST2017
Receiver2:hellomsg:6FriMay1209:08:50CST2017
Receiver1:hellomsg:6FriMay1209:08:50CST2017
Receiver2:hellomsg:7FriMay1209:08:50CST2017
Receiver2:hellomsg:8FriMay1209:08:50CST2017
Receiver2:hellomsg:8FriMay1209:08:50CST2017
Receiver1:hellomsg:7FriMay1209:08:50CST2017
Receiver2:hellomsg:9FriMay1209:08:50CST2017
Receiver2:hellomsg:9FriMay1209:08:50CST2017


和一对多一样，接收端仍然会均匀接收到消息

4、实体类传输

springboot完美的支持对象的发送和接收，不需要格外的配置。

实体类（必须实现序列化接口）：

packagecom.rabbit.user;

importjava.io.Serializable;

publicclassUserimplementsSerializable{
privateStringname;
privateStringpass;
publicStringgetName(){
returnname;
}
publicvoidsetName(Stringname){
this.name=name;
}
publicStringgetPass(){
returnpass;
}
publicvoidsetPass(Stringpass){
this.pass=pass;
}
}


生产者：

packagecom.rabbit.user;

importjava.util.Date;

importorg.springframework.amqp.core.AmqpTemplate;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.stereotype.Component;

@Component
publicclassUserSender{

@Autowired
privateAmqpTemplaterabbitTemplate;

publicvoidsend(){
Useruser=newUser();
user.setName("hzb");
user.setPass("123456789");
System.out.println("usersend:"+user.getName()+"/"+user.getPass());
this.rabbitTemplate.convertAndSend("userQueue",user);
}

}


消费者：

packagecom.rabbit.user;

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;
importorg.springframework.amqp.rabbit.annotation.RabbitListener;
importorg.springframework.stereotype.Component;

@Component
@RabbitListener(queues="userQueue")
publicclassUserReceiver{

@RabbitHandler
publicvoidprocess(Useruser){
System.out.println("userreceive:"+user.getName()+"/"+user.getPass());
}

}


controller:

/**
*实体类传输测试
*/
@PostMapping("/userTest")
publicvoiduserTest(){
userSender.send();
}


用post方式执行：
http://127.0.0.1:8080/rabbit/userTest
结果如下：

usersend:hzb/123456789
userreceive:hzb/123456789


5、topicExChange示例

topic是RabbitMQ中最灵活的一种方式，可以根据binding_key自由的绑定不同的队列

首先对topic规则配置，这里使用两个队列来测试（也就是在Application类中创建和绑定的topic.message和topic.messages两个队列），其中topic.message的bindting_key为

“topic.message”，topic.messages的binding_key为“topic.#”；

生产者：

packagecom.rabbit.topic;

importjava.util.Date;

importorg.springframework.amqp.core.AmqpTemplate;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.stereotype.Component;

@Component
publicclassTopicSender{

@Autowired
privateAmqpTemplaterabbitTemplate;

publicvoidsend(){
Stringmsg1="Iamtopic.mesaagemsg======";
System.out.println("sender1:"+msg1);
this.rabbitTemplate.convertAndSend("exchange","topic.message",msg1);

Stringmsg2="Iamtopic.mesaagesmsg########";
System.out.println("sender2:"+msg2);
this.rabbitTemplate.convertAndSend("exchange","topic.messages",msg2);
}

}


消费者1（topic.message）

packagecom.rabbit.topic;

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;
importorg.springframework.amqp.rabbit.annotation.RabbitListener;
importorg.springframework.stereotype.Component;

@Component
@RabbitListener(queues="topic.message")
publicclasstopicMessageReceiver{

@RabbitHandler
publicvoidprocess(Stringmsg){
System.out.println("topicMessageReceiver:"+msg);
}

}


消费者２（topic.messages）

packagecom.rabbit.topic;

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;
importorg.springframework.amqp.rabbit.annotation.RabbitListener;
importorg.springframework.stereotype.Component;

@Component
@RabbitListener(queues="topic.messages")
publicclasstopicMessagesReceiver{

@RabbitHandler
publicvoidprocess(Stringmsg){
System.out.println("topicMessagesReceiver:"+msg);
}

}


controller:

/**
*topicexchange类型rabbitmq测试
*/
@PostMapping("/topicTest")
publicvoidtopicTest(){
topicSender.send();
}


用post方式执行：
http://127.0.0.1:8080/rabbit/topicTest
结果如下：

sender1:Iamtopic.mesaagemsg======
sender2:Iamtopic.mesaagesmsg########
topicMessageReceiver:Iamtopic.mesaagemsg======
topicMessagesReceiver:Iamtopic.mesaagemsg======
topicMessagesReceiver:Iamtopic.mesaagesmsg########


由以上结果可知：sender1发送的消息,routing_key是“topic.message”，所以exchange里面的绑定的binding_key是“topic.message”，topic.＃都符合路由规则;所以sender1

发送的消息，两个队列都能接收到；

sender2发送的消息，routing_key是“topic.messages”，所以exchange里面的绑定的binding_key只有topic.＃都符合路由规则;所以sender2发送的消息只有队列

topic.messages能收到。

6、fanoutExChange示例

Fanout就是我们熟悉的广播模式或者订阅模式，给Fanout转发器发送消息，绑定了这个转发器的所有队列都收到这个消息。

这里使用三个队列来测试（也就是在Application类中创建和绑定的fanout.A、fanout.B、fanout.C）这三个队列都和Application中创建的fanoutExchange转发器绑定。

生产者：

packagecom.rabbit.fanout;

importjava.util.Date;

importorg.springframework.amqp.core.AmqpTemplate;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.stereotype.Component;

@Component
publicclassFanoutSender{

@Autowired
privateAmqpTemplaterabbitTemplate;

publicvoidsend(){
StringmsgString="fanoutSender:helloiamhzb";
System.out.println(msgString);
this.rabbitTemplate.convertAndSend("fanoutExchange","abcd.ee",msgString);
}

}


消费者A:

packagecom.rabbit.fanout;

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;
importorg.springframework.amqp.rabbit.annotation.RabbitListener;
importorg.springframework.stereotype.Component;

@Component
@RabbitListener(queues="fanout.A")
publicclassFanoutReceiverA{

@RabbitHandler
publicvoidprocess(Stringmsg){
System.out.println("FanoutReceiverA:"+msg);
}

}


消费者Ｂ：

packagecom.rabbit.fanout;

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;
importorg.springframework.amqp.rabbit.annotation.RabbitListener;
importorg.springframework.stereotype.Component;

@Component
@RabbitListener(queues="fanout.B")
publicclassFanoutReceiverB{

@RabbitHandler
publicvoidprocess(Stringmsg){
System.out.println("FanoutReceiverB:"+msg);
}

}


消费者C:

packagecom.rabbit.fanout;

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;
importorg.springframework.amqp.rabbit.annotation.RabbitListener;
importorg.springframework.stereotype.Component;

@Component
@RabbitListener(queues="fanout.C")
publicclassFanoutReceiverC{

@RabbitHandler
publicvoidprocess(Stringmsg){
System.out.println("FanoutReceiverC:"+msg);
}

}


controller:

/**
*fanoutexchange类型rabbitmq测试
*/
@PostMapping("/fanoutTest")
publicvoidfanoutTest(){
fanoutSender.send();
}


用post方式执行：
http://127.0.0.1:8080/rabbit/fanoutTest
结果如下：

fanoutSender:helloiamhzb
FanoutReceiverC:fanoutSender:helloiamhzb
FanoutReceiverB:fanoutSender:helloiamhzb
FanoutReceiverA:fanoutSender:helloiamhzb


由以上结果可知：就算fanoutSender发送消息的时候，指定了routing_key为"abcd.ee"，但是所有接收者都接受到了消息

7、带callback的消息发送

增加回调处理，这里不再使用application.properties默认配置的方式，会在程序中显示的使用文件中的配置信息。该示例中没有新建队列和exchange，用的是第5节中的topic.messages队列和exchange转发器。消费者也是第5节中的topicMessagesReceiver

rabbitmq配置类：

packagecom.rabbit.callback;

importorg.springframework.amqp.rabbit.connection.CachingConnectionFactory;
importorg.springframework.amqp.rabbit.connection.ConnectionFactory;
importorg.springframework.amqp.rabbit.core.RabbitTemplate;
importorg.springframework.beans.factory.annotation.Value;
importorg.springframework.beans.factory.config.ConfigurableBeanFactory;
importorg.springframework.context.annotation.Bean;
importorg.springframework.context.annotation.Scope;

publicclassRabbitConfig{

@Value("${spring.rabbitmq.host}")
privateStringaddresses;

@Value("${spring.rabbitmq.port}")
privateStringport;

@Value("${spring.rabbitmq.username}")
privateStringusername;

@Value("${spring.rabbitmq.password}")
privateStringpassword;

@Value("${spring.rabbitmq.virtual-host}")
privateStringvirtualHost;

@Value("${spring.rabbitmq.publisher-confirms}")
privatebooleanpublisherConfirms;

@Bean
publicConnectionFactoryconnectionFactory(){

CachingConnectionFactoryconnectionFactory=newCachingConnectionFactory();
connectionFactory.setAddresses(addresses+":"+port);
connectionFactory.setUsername(username);
connectionFactory.setPassword(password);
connectionFactory.setVirtualHost(virtualHost);
/**如果要进行消息回调，则这里必须要设置为true*/
connectionFactory.setPublisherConfirms(publisherConfirms);
returnconnectionFactory;
}

@Bean
/**因为要设置回调类，所以应是prototype类型，如果是singleton类型，则回调类为最后一次设置*/
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
publicRabbitTemplaterabbitTemplatenew(){
RabbitTemplatetemplate=newRabbitTemplate(connectionFactory());
returntemplate;
}

}


生产者：

packagecom.rabbit.callback;

importjava.util.Date;
importjava.util.UUID;

importorg.springframework.amqp.rabbit.core.RabbitTemplate;
importorg.springframework.amqp.rabbit.support.CorrelationData;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.stereotype.Component;

@Component
publicclassCallBackSenderimplementsRabbitTemplate.ConfirmCallback{
@Autowired
privateRabbitTemplaterabbitTemplatenew;
publicvoidsend(){

rabbitTemplatenew.setConfirmCallback(this);
Stringmsg="callbackSender:iamcallbacksender";
System.out.println(msg);
CorrelationDatacorrelationData=newCorrelationData(UUID.randomUUID().toString());
System.out.println("callbackSenderUUID:"+correlationData.getId());
this.rabbitTemplatenew.convertAndSend("exchange","topic.messages",msg,correlationData);
}

publicvoidconfirm(CorrelationDatacorrelationData,booleanack,Stringcause){
//TODOAuto-generatedmethodstub
System.out.println("callbakckconfirm:"+correlationData.getId());
}
}


消费者：第5节中的topicMessagesReceiver

controller:

@PostMapping("/callback")
publicvoidcallbak(){
callBackSender.send();
}


用post方式执行：
http://127.0.0.1:8080/rabbit/callback
结果如下：

callbackSender:iamcallbacksender
callbackSenderUUID:cd0c80a6-4c65-4bf9-b4f8-f3b1180755d6
callbakckconfirm:cd0c80a6-4c65-4bf9-b4f8-f3b1180755d6
topicMessagesReceiver:callbackSender:iamcallbacksender


从上面可以看出callbackSender发出的UUID，收到了回应，又传回来了。
                                            

                                        

                        
                        内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 点击这里给我发消息 
                    

                    

                    

                        

                         标签： 
                                                

                        

                    


                

            


            

                
相关文章推荐

                

                
            

        

        

            

            
            

                

                    
新的分享

                    

                        
                    

                

            


            

                

                    
章节导航