Spring Boot集成RabbitMQ相比于Spring集成RabbitMQ简单很多,有兴趣了解Spring集成RabbitMQ的同学可以看我之前的《RabbitMQ学习笔记》系列的博客,特别是优先级消息队列,这里就不做重复介绍了。
在之前创建的Spring Boot项目(Spring Boot实战一:搭建Spring Boot开发环境)中加入依赖jar和RabbitMQ的配置(这里使用的是本地安装的RabbitMQ,需要安装的同学,可以参考我之前的博客:RabbitMQ学习笔记一:本地Windows环境安装RabbitMQ Server):
在pom.xml中加入依赖jar:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
在application.properties中加入RabbitMQ的配置信息:
# RabbitMQ config
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtualHost=/
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.template.mandatory=true
生产者SenderService:
package com.example.demo.service;
import javax.annotation.PostConstruct;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class SenderService implements RabbitTemplate.ConfirmCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct
public void init()
{
rabbitTemplate.setConfirmCallback(this);
}
public void send(String msg)
{
// rabbitTemplate.convertAndSend("queue", msg); // 使用这个方法,则对应的消费者Listener中的process方法不能有返回值,否则会报错
Object receive = rabbitTemplate.convertSendAndReceive("queue", msg);
System.out.println("receive " + receive);
}
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause)
{
if (ack) {
System.out.println("消息发送确认成功");
} else {
System.out.println("消息发送确认失败:" + cause);
}
}
}
消费者Listener:
package com.example.demo.service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class Listener {
private static final Logger logger = LoggerFactory.getLogger(Listener.class);
@RabbitListener(queues = "queue")
public String process(String msg)
{
logger.info("Listener: " + msg);
return msg;
}
}
控制器RabbitmqController:
package com.example.demo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import com.example.demo.service.SenderService;
@RestController
public class RabbitmqController {
@Autowired
private SenderService sender;
@GetMapping("/send")
public String send(String msg)
{
sender.send(msg);
return "Send OK.";
}
}
启动访问:http://localhost:8080/send?msg=rabbitmq ,打印结果如下:
2017-08-14 14:15:34.588 INFO 6452 --- [nio-8080-exec-1] o.a.c.c.C.[Tomcat].[localhost].[/] : Initializing Spring FrameworkServlet 'dispatcherServlet'
2017-08-14 14:15:34.589 INFO 6452 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet : FrameworkServlet 'dispatcherServlet': initialization started
2017-08-14 14:15:34.614 INFO 6452 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet : FrameworkServlet 'dispatcherServlet': initialization completed in 25 ms
消息发送确认成功
2017-08-14 14:15:34.699 INFO 6452 --- [cTaskExecutor-1] com.example.demo.service.Listener : Listener: rabbitmq
receive rabbitmq
PS:需要在RabbitMQ Server创建对应的队列,不了解RabbitMQ的可以参考我之前的《RabbitMQ学习笔记》系列的博客。