1.pom添加配置
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>2.application.properties文件添加配置
spring.rabbitmq.host=xxxx
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
#链接的虚拟主机
spring.rabbitmq.virtual-host=/
#手动确认模式，消费者调用ack方法后，RabbitMQ确认消息。
spring.rabbitmq.listener.direct.acknowledge-mode=manual
#交换机
mq.stock.exchange=stock.event.exchange
#普通队列
mq.stock.queue=event.sales-agency.stock.release.queue
#普通键
mq.stock.routingKey=stock.ordinary.key
#延时队列
mq.stock.delay-queue=event.sales-agency.stock.delay.queue
#延时键
mq.stock.delay-routing-key=stock.delay.key
#死信队列
mq.stock.die-queue=event.sales-agency.stock.die.queue
#死信队列键
mq.stock.die-key=stock.die.key3.获取配置文件信息
package com.laoli.grocery.rabbitmq;
import lombok.Getter;
import lombok.Setter;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
@Getter
@Setter
@Configuration
//获取配置文件中的mq开头的属性
@ConfigurationProperties(prefix = "mq")
public class MqProperties {
private MqConfig stock = new MqConfig();
@Getter
@Setter
public static final class MqConfig {
// 交换机
private String exchange;
//普通队列
private String queue;
//普通队列key
private String routingKey;
//延时队列
private String delayQueue;
//延时队列key
private String delayRoutingKey;
//死信队列
private String dieQueue;
//死信队列key
private String dieKey;
}
}
4.mq配置文件
package com.laoli.grocery.rabbitmq;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class RabbitConfig {
@Resource
private MqProperties mq;
/**
* 交换机
* @return
*/
@Bean
public Exchange stockEventExchange() {
return ExchangeBuilder.directExchange(mq.getStock().getExchange()).durable(true).build();
}
/**
* 普通队列
* @return
*/
@Bean
public Queue stockReleaseQueue() {
return QueueBuilder.durable(mq.getStock().getQueue()).build();
}
/**
* 普通队列绑定交换机
* @param exchange 交换机
* @param queue 队列
* mq.getStock().getRoutingKey() key
* @return
*/
@Bean
public Binding bindingReleaseQueue(@Qualifier("stockEventExchange") Exchange exchange, @Qualifier("stockReleaseQueue") Queue queue) {
//主题RoutingKey要和消费者的RoutingKey 一致
return BindingBuilder.bind(queue).to(exchange).with(mq.getStock().getRoutingKey()).noargs();
}
/**
* 死信队列
* @return
*/
@Bean
public Queue stockDieReleaseQueue() {
return QueueBuilder.durable(mq.getStock().getDieQueue()).build();
}
/**
* 死信队列绑定交换机
* @param exchange 交换机
* @param queue 队列
* mq.getStock().getRoutingKey() key
* @return
*/
@Bean
public Binding bindingDieQueue(@Qualifier("stockEventExchange") Exchange exchange, @Qualifier("stockDieReleaseQueue") Queue queue) {
return BindingBuilder.bind(queue).to(exchange).with(mq.getStock().getDieKey()).noargs();
}
/**
* 创建延时队列
* @return
*/
@Bean
public Queue stockDelayQueue() {
// 创建队列需要的参数
Map<String, Object> args = new HashMap<>();
// 设置TTL，单位毫秒，即下列为15秒，延迟队列过期时间为15秒
args.put("x-message-ttl", 15000);
// 设置死信的目的交换机
args.put("x-dead-letter-exchange", mq.getStock().getExchange());
// 设置死信交给目的交换机时的路由键
args.put("x-dead-letter-routing-key", mq.getStock().getDieKey());
return QueueBuilder.durable(mq.getStock().getDelayQueue()).withArguments(args).build();
}
/**
* 延时队列
* 将交换机和队列绑定
* @param exchange 交换机
* @param queue 队列
* @return
*/
@Bean
public Binding bindingDelayQueue(@Qualifier("stockEventExchange") Exchange exchange, @Qualifier("stockDelayQueue") Queue queue) {
return BindingBuilder.bind(queue).to(exchange).with(mq.getStock().getDelayRoutingKey()).noargs();
}
}
5.mq生产者，以下是为了方便测试
package com.laoli.grocery.rabbitmq.controller;
import com.laoli.grocery.rabbitmq.MqProperties;
import com.laoli.grocery.util.BaseController;
import com.laoli.grocery.util.Result;
import com.laoli.grocery.util.TimeUtils;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
@RestController
@RequestMapping("/rabbitmq")
public class MqController extends BaseController {
@Resource
private MqProperties mq;
@Resource
private RabbitTemplate rabbitTemplate;
@GetMapping(value = "/sendMsg")
public Result sendMsg(@RequestParam String message) {
System.out.println("普通队列消息发送时间: "+TimeUtils.getTime());
rabbitTemplate.convertAndSend(mq.getStock().getExchange(),
mq.getStock().getRoutingKey(), message);
return success();
}
@GetMapping(value = "/sendDelayMsg")
public Result sendDelayMsg(@RequestParam String message) {
System.out.println("延时队列列消息发送时间: "+TimeUtils.getTime());
rabbitTemplate.convertAndSend(mq.getStock().getExchange(),
mq.getStock().getDelayRoutingKey(), message);
return success();
}
}
6.mq消费者
package com.laoli.grocery.rabbitmq;
import com.laoli.grocery.util.TimeUtils;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
@Slf4j
@Component
public class ReceiverMq {
/**
* 获取队列里面的消息
* @param message
* @param data
* @param channel
* @throws IOException
*/
@RabbitListener(queues = "${mq.stock.queue}")
public void getMessage(Message message, Object data, Channel channel) throws IOException {
System.out.println("普通队列收到的时间:【"+TimeUtils.getTime()+"】");
System.out.println("普通队列接收到message：【"+message+"】");
System.out.println("普通队列接收到data：【"+data+"】");
}
@RabbitListener(queues = "${mq.stock.die-queue}")
public void getDieMessage(Message message, Object data, Channel channel) throws IOException {
System.out.println("死信队列收到的时间:【"+TimeUtils.getTime()+"】");
System.out.println("死信队列接收到message：【"+message+"】");
System.out.println("死信队列接收到data：【"+data+"】");
}
}
7.时间工具类
package com.laoli.grocery.util;
import java.text.SimpleDateFormat;
import java.util.Date;
public class TimeUtils {
public static String getTime() {
Date dt = new Date();
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
return sdf.format(dt);
}
}
