3.10
1.在opt文件夹里建一个rabbitmq文件夹,把插件放进去
2.拷贝到容器内plugins目录下(rabbitmq是容器的name,也可以使用容器id)
docker cp /opt/rabbitmq/rabbitmq_delayed_message_exchange-3.10.0.ez rabbitmq:/plugins
进入 Docker 容器
docker exec -it rabbitmq /bin/bash
在plugins内启用插件
#先执行,解除防火墙限制,增加文件权限
umask 0022
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
退出容器
exit
重启 RabbitMQ
docker restart rabbitmq
通过UI查看
原理
代码实现,
消费者:自定义了一个延迟交换机,因为spring没有封装,在map集合中指定哪个交换机模版
@Configuration
public class DlelayConsumer {
@Bean //自定义延迟交换机
public CustomExchange exchange() {
HashMap<String, Object> args = MapUtil.of("x-delayed-type", "fanout");
//指定交换机类型
return new CustomExchange("dly_ordering_ok", "x-delayed-message", true,
false, args);
}
@Bean
public Queue queue() {
return QueueBuilder.durable("dly_ordering_ok_wms").build();
}
@Bean
public Binding binding() {
return BindingBuilder.bind(queue()).to(exchange()).with("").noargs();
}
//@RabbitListener(queues = "dly_ordering_ok_wms")
public void consume(OrderingOk msg) throws IOException {
System.out.println("ordering_ok_wms:"+msg);
}
}
生产者:最后一个参数是给信息设置延迟时间和持久化,.默认是持久化的;
@Service
public class DelayProvider {
@Autowired
private RabbitTemplate rabbitTemplate;
public void send(OrderingOk msg) {
rabbitTemplate.convertAndSend("dly_ordering_ok", "", msg, message -> {
Long id = msg.getId();
int delay = 0;
switch (id.intValue()){
case 1:
delay = 50*1000;
break;
case 2:
delay = 40*1000;
break;
case 3:
delay = 30*1000;
break;
case 4:
delay = 20*1000;
break;
case 5:
delay = 10*1000;
break;
}
message.getMessageProperties().setDelay(delay);//设置延迟时间
//如果RabbitMQ成功将消息写入磁盘,并且至少被一个交换器接受,那么消息就被认为是持久化的 //设置消息持久化,默认是PERSISTENT
message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
return message;
});
}
}
测试:写一个对象让生产者去发送
@SpringBootTest
public class DelayTests {
@Autowired
private DelayProvider delayProvider;
@Test
void test() throws IOException {
for (int i = 1; i < 6; i++) {
OrderingOk orderingOk = new OrderingOk().setId((long) i).setUserName("张三").setOrderItems(CollUtil.newArrayList(new OrderItem().
setId(1).setProductId(1).setProductName("苹果").setPrice(BigDecimal.valueOf(10.0)).setQty(12)));
delayProvider.send(orderingOk);
}
System.out.println("发送成功");
System.in.read();
}
}
#当设置为 CORRELATED 时,Spring Boot会为每个发布的消息生成一个唯一的关联ID,并将这些ID与确认信息相关联。这样,当消息被确认或未被确认时,可以通过关联ID追踪到具体的每条消息。
spring.rabbitmq.publisher-confirm-type = CORRELATED
#当设置为 true 时,如果消息无法被路由到任何一个队列(例如,由于队列不存在或消息的路由键无效),那么RabbitMQ会将这些消息退回给生产者。 这允许生产者处理这些未能正确路由的消息。
spring.rabbitmq.publisher-returns = true