导入依赖
一、简单整合org.springframework.boot spring-boot-starter-amqp
我们这里在两个项目中完成一个生产者项目，一个消费者项目
生产者
配置文件
server.port=8080 spring.rabbitmq.host=192.168.195.157 spring.rabbitmq.port=5672 spring.rabbitmq.virtual-host=/ spring.rabbitmq.username=guest spring.rabbitmq.password=guest
controller（发送端）
@RestController
public class SendController {
@Autowired
RabbitTemplate rabbitTemplate;
@RequestMapping("/sendmail")
public String sendmail(){
//发送消息
rabbitTemplate.convertAndSend("exchangetopic","email","email");
return "sendmail";
}
@RequestMapping("/sendsms")
public String sendSms(){
//发送消息
rabbitTemplate.convertAndSend("exchangetopic","sms","sms");
return "sendsms";
}
@RequestMapping("/sendall")
public String sendAll(){
//发送消息
rabbitTemplate.convertAndSend("exchangetopic","sms.email","sms.email");
return "sendall";
}
}
消费者
配置文件
server.port=8081 spring.rabbitmq.host=192.168.195.157 spring.rabbitmq.port=5672 spring.rabbitmq.virtual-host=/ spring.rabbitmq.username=guest spring.rabbitmq.password=guest
配置类
@Configuration
public class RabbitmqConfiguration {
//定义交换机
@Bean
public TopicExchange exchangeTopic() {
return new TopicExchange("exchangetopic");
}
//定义队列
@Bean
public Queue queueMail() {
return new Queue("queueemail");
}
@Bean
public Queue queueMsg() {
return new Queue("queuesms");
}
@Bean
public Binding bindingEmail(TopicExchange exchange, Queue queueMail) {
return BindingBuilder.bind(queueMail).to(exchange).with("#.email.#");
}
@Bean
public Binding bindingMsg(TopicExchange exchange, Queue queueMsg) {
return BindingBuilder.bind(queueMsg).to(exchange).with("#.sms.#");
}
}
消费者处理请求
@Component
public class MyConsumer {
//指定当前方法监听那个队列
//String msg: 消息体
//Message: 消息对象
//Channel: 信道
@RabbitListener(queues = "queueemail")
public void receiveEmail(String msg, Message message, Channel channel) {
System.out.println("处理Email");
System.out.println("msg:" + msg);
System.out.println("message:" + message);
System.out.println("channel:" + channel);
System.out.println("---------------");
}
@RabbitListener(queues = "queuesms")
public void receiveSms(String msg, Message message, Channel channel) {
System.out.println("处理Sms");
System.out.println("msg:" + msg);
System.out.println("message:" + message);
System.out.println("channel:" + channel);
System.out.println("---------------");
}
}
执行结果：
配置文件
server.port=8080 spring.rabbitmq.host=192.168.195.157 spring.rabbitmq.port=5672 spring.rabbitmq.virtual-host=/ spring.rabbitmq.username=guest spring.rabbitmq.password=guest #生产者confrim #spring.rabbitmq.publisher-/confirm/is=true spring.rabbitmq.publisher-/confirm/i-type=correlated #生产者return spring.rabbitmq.publisher-returns=true
添加配置类开启监听
package com.acoffee.configuration;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import javax.annotation.PostConstruct;
@Configuration
public class CallbackConfiguration implements RabbitTemplate./confirm/iCallback, RabbitTemplate.ReturnsCallback {
@Autowired
RabbitTemplate rabbitTemplate;
//@PostConstruct @PreDestroy：类似生命周期函数
@PostConstruct
public void setUp(){
rabbitTemplate.set/confirm/iCallback(this);
rabbitTemplate.setReturnsCallback(this);
}
@Override
public void /confirm/i(CorrelationData correlationData, boolean ack, String cause) {
System.out.println("/confirm/i");
System.out.println("correlationdata:"+correlationData);
System.out.println("ack:"+ack);
System.out.println("cause:"+cause);
System.out.println("-----------------");
}
@Override
public void returnedMessage(ReturnedMessage returned) {
System.out.println("Return");
System.out.println("message:"+returned.getMessage());
System.out.println("replyText:"+returned.getReplyText());
System.out.println("exchange:"+returned.getExchange());
System.out.println("routingKey"+returned.getRoutingKey());
System.out.println("-----------------");
}
}
执行结果：
我们测试了三种
第一种：故意将exchange的名字写错，可以看出返回的 ack 为 false 就是我们说的 nack，并且从cause中就可以看出错误原因，我们原来也说过在基础api中是看不到nack的，在springboot中是可以的。
第二种：故意写错routingKey，从结果可以看出的ack为true，说明的已经到交换机了，但是由于通过routingKey找不到对应的管道所以给我们返回了消息，在replyText中也可以看到replyText:NO_ROUTE造成的
第三种：正确情况
只返回了/confirm/i中的信息，可以看出ack为true，并且消费者也接收并处理了消息
默认就是自动确认，一下全取出来，宕机就会出现问题
手动确认+限流消费者配置文件
server.port=8081 spring.rabbitmq.host=192.168.195.157 spring.rabbitmq.port=5672 spring.rabbitmq.virtual-host=/ spring.rabbitmq.username=guest spring.rabbitmq.password=guest #消费者手动确认 spring.rabbitmq.listener.simple.acknowledge-mode=manual #限流 spring.rabbitmq.listener.simple.prefetch=1
@RabbitListener(queues = "queueemail")
public void receiveEmail(String msg, Message message, Channel channel) throws Exception {
System.out.println("处理Email");
System.out.println("msg:" + msg);
System.out.println("message:" + message);
System.out.println("channel:" + channel);
//手动确认
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
System.out.println("---------------");
}
执行结果：
一个一个读取，一个一个确认
@RequestMapping("/sendmail")
public String sendmail() {
//发送消息
rabbitTemplate.convertAndSend("exchangetopic", "email", "email1",
//设置消息属性
new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setExpiration("10000");
//message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
message.getMessageProperties().getHeaders().put("h1", "v1");
message.getMessageProperties().getHeaders().put("h2", "v2");
return message;
}
});
return "sendmail";
}
执行结果：
配置类中添加JSON消息转化器，底层默认使用的Jackson
@Bean
public MessageConverter messageConverter(){
return new Jackson2JsonMessageConverter();
}
消费者和生产者都要配置实体类，且包名和类名必须一样
生产者
@RequestMapping("/sendsms")
public String sendSms() {
//发送消息:传对象
rabbitTemplate.convertAndSend("exchangetopic", "sms", new Student(1,"tom"));
return "sendsms";
}
消费者
@RabbitListener(queues = "queuesms")
public void receiveSms(Student msg, Message message, Channel channel) throws IOException {
System.out.println("处理Sms");
System.out.println("msg:" + msg);
System.out.println("message:" + message);
System.out.println("channel:" + channel);
//手动确认
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
System.out.println("---------------");
}
执行结果：
生产者
@RequestMapping("/sendorder")
public String sendOrder(){
//发送消息
System.out.println("订单产生时间"+new Date());
rabbitTemplate.convertAndSend("exchangeorder","order","order--info");
return "sendorder";
}
消费者
配置模拟队列以及死信队列
@Bean
public TopicExchange exchangeOrder(){
return new TopicExchange("exchangeorder");
}
@Bean
public Queue queueorder(){
HashMap args = new HashMap<>();
//10秒过期，直接进入指定的死信队列
args.put("x-message-ttl",10000);
args.put("x-dead-letter-exchange","exchangeorderdlx");
args.put("x-dead-letter-routing-key","orderdlx");
Queue queueorder = new Queue("queueorder", true, false, false, args);
return queueorder;
}
@Bean
public Binding bindingqueueorder(Queue queueorder,TopicExchange exchangeorder){
return BindingBuilder.bind(queueorder).to(exchangeorder).with("#.order.#");
}
@Bean
public TopicExchange exchangeorderdlx(){
return new TopicExchange("exchangeorderdlx");
}
@Bean
public Queue queueorderdlx(){
Queue queueorderdlx = new Queue("queueorderdlx");
return queueorderdlx;
}
@Bean
public Binding bindingqueueorderdlx(Queue queueorderdlx,TopicExchange exchangeorderdlx){
return BindingBuilder.bind(queueorderdlx).to(exchangeorderdlx).with("#.orderdlx.#");
}
在死信队列中处理消息
@RabbitListener(queues = "queueorderdlx")
public void receiveOrdereddlx(String msg, Message message, Channel channel) throws IOException {
System.out.println("处理queueorderdlx");
System.out.println("msg:" + msg);
System.out.println("message:" + message);
System.out.println("channel:" + channel);
//手动确认
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
System.out.println("---------------");
}
七、延迟队列
安装延迟插件
下载地址
我用的是自己的包，这里需要注意版本兼容问题。
传到linux上去，然后安装解压软件
yum install zip yum install -y unzip
解压
unzip 要解压的文件 -d 解压的目标位置 unzip rabbitmq_delayed_message_exchange-20171201-3.7.x.zip -d /usr/lib/rabbitmq/lib/rabbitmq_server-3.7.16/plugins/
查看插件列表
rabbitmq-plugins list
安装
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
重启
systemctl restart rabbitmq-server
启用延迟队列：发送消息，立即发送到交换机，进入交换以后到了指定延迟时间，才从交换机发送的队列
交换机启用
@Bean
public TopicExchange exchangedelay(){
TopicExchange exchangedelay = new TopicExchange("exchangedelay");
//交换机启用延迟
exchangedelay.setDelayed(true);
return exchangedelay;
}
@Bean
public Queue queuedelay(){
Queue queuedelay = new Queue("queuedelay");
return queuedelay;
}
@Bean
public Binding bindingqueuedelay(Queue queuedelay,TopicExchange exchangedelay){
return BindingBuilder.bind(queuedelay).to(exchangedelay).with("#.delay.#");
}
消息属性设置
@RequestMapping("/senddelay")
public String sendDelay(){
//发送消息
System.out.println("延迟队列发送时间:"+new Date());
rabbitTemplate.convertAndSend("exchangedelay", "delay", "delay--info",
new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
//延迟十秒进入队列
message.getMessageProperties().setDelay(10000);
return message;
}
}
);
return "senddelay";
}
执行结果：
从结果我们可以看出它先是报没有路由，因为他在转换机中等十秒才会进入队列，而服务器会直接响应回return信息，等十秒就会看到结果，如果不想看，这里我们可以做一些判断，避免这种信息
依赖
org.springframework.boot spring-boot-starter-web org.projectlombok lombok true org.springframework.boot spring-boot-starter-test test org.springframework.boot spring-boot-starter-amqp
生产者
server.port=8080 spring.rabbitmq.host=192.168.195.157 spring.rabbitmq.port=5672 spring.rabbitmq.virtual-host=/ spring.rabbitmq.username=guest spring.rabbitmq.password=guest #生产者confrim #spring.rabbitmq.publisher-/confirm/is=true spring.rabbitmq.publisher-/confirm/i-type=correlated #生产者return spring.rabbitmq.publisher-returns=true #解决循环依赖问题2.6.x可能会出现 spring.main.allow-circular-references=true
实体类
@Data
@NoArgsConstructor
@AllArgsConstructor
public class Student implements Serializable {
int id;
String name;
}
配置类
@Configuration
public class CallbackConfiguration implements RabbitTemplate./confirm/iCallback, RabbitTemplate.ReturnsCallback {
@Autowired
RabbitTemplate rabbitTemplate;
//@PostConstruct @PreDestroy：类似生命周期函数
@PostConstruct
public void setUp(){
rabbitTemplate.set/confirm/iCallback(this);
rabbitTemplate.setReturnsCallback(this);
}
@Override
public void /confirm/i(CorrelationData correlationData, boolean ack, String cause) {
System.out.println("/confirm/i");
System.out.println("correlationdata:"+correlationData);
System.out.println("ack:"+ack);
System.out.println("cause:"+cause);
System.out.println("-----------------");
}
@Override
public void returnedMessage(ReturnedMessage returned) {
System.out.println("Return");
System.out.println("message:"+returned.getMessage());
System.out.println("replyText:"+returned.getReplyText());
System.out.println("exchange:"+returned.getExchange());
System.out.println("routingKey"+returned.getRoutingKey());
System.out.println("-----------------");
}
@Bean
public MessageConverter messageConverter(){
return new Jackson2JsonMessageConverter();
}
}
controller层(发送请求)
import com.acoffee.entity.Student;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.Date;
@RestController
public class SendController {
@Autowired
RabbitTemplate rabbitTemplate;
@RequestMapping("/sendmail")
public String sendmail() {
//发送消息
for (int i = 1; i <= 10; i++) {
rabbitTemplate.convertAndSend("exchangetopic", "email", "email" + i,
//设置消息属性
new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
//过期时间
message.getMessageProperties().setExpiration("10000");
//message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
message.getMessageProperties().getHeaders().put("h1", "v1");
message.getMessageProperties().getHeaders().put("h2", "v2");
return message;
}
});
}
return "sendmail";
}
@RequestMapping("/sendsms")
public String sendSms() {
//发送消息:传对象
rabbitTemplate.convertAndSend("exchangetopic", "sms", new Student(1,"tom"));
return "sendsms";
}
@RequestMapping("/sendall")
public String sendAll() {
//发送消息
rabbitTemplate.convertAndSend("exchangetopic", "sms.email", "sms.email");
return "sendall";
}
@RequestMapping("/sendorder")
public String sendOrder(){
//发送消息
System.out.println("订单产生时间"+new Date());
rabbitTemplate.convertAndSend("exchangeorder","order","order--info");
return "sendorder";
}
@RequestMapping("/senddelay")
public String sendDelay(){
//发送消息
System.out.println("延迟队列发送时间:"+new Date());
rabbitTemplate.convertAndSend("exchangedelay", "delay", "delay--info",
new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setDelay(10000);
return message;
}
}
);
return "senddelay";
}
}
消费者
实体类
@Data
@NoArgsConstructor
@AllArgsConstructor
public class Student implements Serializable {
int id;
String name;
}
配置类
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
@Configuration
public class RabbitmqConfiguration {
//定义交换机
@Bean
public TopicExchange exchangeTopic() {
return new TopicExchange("exchangetopic");
}
//定义队列
@Bean
public Queue queueMail() {
return new Queue("queueemail");
}
@Bean
public Queue queueMsg() {
return new Queue("queuesms");
}
@Bean
public Binding bindingEmail(TopicExchange exchangeTopic, Queue queueMail) {
return BindingBuilder.bind(queueMail).to(exchangeTopic).with("#.email.#");
}
@Bean
public Binding bindingMsg(TopicExchange exchangeTopic, Queue queueMsg) {
return BindingBuilder.bind(queueMsg).to(exchangeTopic).with("#.sms.#");
}
//对象转json
@Bean
public MessageConverter messageConverter(){
return new Jackson2JsonMessageConverter();
}
@Bean
public TopicExchange exchangeOrder(){
return new TopicExchange("exchangeorder");
}
@Bean
public Queue queueorder(){
HashMap args = new HashMap<>();
args.put("x-message-ttl",10000);
args.put("x-dead-letter-exchange","exchangeorderdlx");
args.put("x-dead-letter-routing-key","orderdlx");
Queue queueorder = new Queue("queueorder", true, false, false, args);
return queueorder;
}
@Bean
public Binding bindingqueueorder(Queue queueorder,TopicExchange exchangeOrder){
return BindingBuilder.bind(queueorder).to(exchangeOrder).with("#.order.#");
}
@Bean
public TopicExchange exchangeorderdlx(){
return new TopicExchange("exchangeorderdlx");
}
@Bean
public Queue queueorderdlx(){
Queue queueorderdlx = new Queue("queueorderdlx");
return queueorderdlx;
}
@Bean
public Binding bindingqueueorderdlx(Queue queueorderdlx,TopicExchange exchangeorderdlx){
return BindingBuilder.bind(queueorderdlx).to(exchangeorderdlx).with("#.orderdlx.#");
}
//延迟队列
@Bean
public TopicExchange exchangedelay(){
TopicExchange exchangedelay = new TopicExchange("exchangedelay");
//交换机启用延迟
exchangedelay.setDelayed(true);
return exchangedelay;
}
@Bean
public Queue queuedelay(){
Queue queuedelay = new Queue("queuedelay");
return queuedelay;
}
@Bean
public Binding bindingqueuedelay(Queue queuedelay,TopicExchange exchangedelay){
return BindingBuilder.bind(queuedelay).to(exchangedelay).with("#.delay.#");
}
}
MyConsumer 类
@Component
public class MyConsumer {
//指定当前方法监听那个队列
//String msg: 消息体
//Message: 消息对象
//Channel: 信道
@RabbitListener(queues = "queueemail")
public void receiveEmail(String msg, Message message, Channel channel) throws Exception {
Thread.sleep(10000);
System.out.println("处理Email");
System.out.println("msg:" + msg);
System.out.println("message:" + message);
System.out.println("channel:" + channel);
//手动确认
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
System.out.println("---------------");
}
@RabbitListener(queues = "queuesms")
public void receiveSms(Student msg, Message message, Channel channel) throws IOException {
System.out.println("处理Sms");
System.out.println("msg:" + msg);
System.out.println("message:" + message);
System.out.println("channel:" + channel);
//手动确认
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
System.out.println("---------------");
}
}
