Ribbit Mq 实现延迟消息

--------------------好记性不如烂笔头---------------------------

windows 环境,使用 rabbit Mq 需要安装， erl 和 rabbit Mq

1.erl 安装完需要配置环境变量

2.查询 erl 是否安装好，cmd-->erl -version

erl -version
3.MQ 安装目录下
D:\anzhuang\rabbitmq_server-3.8.9\sbin
启动：cmd-->rabbitmq-server

访问：http://localhost:15672/#/

guest/guest （用户名/密码）

--以上可以登录，进行下面操作，配置文件

spring:


rabbitmq:
  host: localhost
  port: 5672
  virtualHost: /
  username : guest
  password : guest
  listener:
    simple:
      acknowledge-mode: manual

pom文件

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置类

package com.example.servicebuy.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
* Created by qiuzhijie.
* Date: 2019-01-07
* 备注: mq config information
*/
@Configuration
public class RabbitMQConfig {

private Logger log = LoggerFactory.getLogger(this.getClass());
@Value("${spring.rabbitmq.host}")
private String HOST;
@Value("${spring.rabbitmq.port}")
private Integer PORT;
@Value("${spring.rabbitmq.virtualHost}")
private String VIRTUALHOST;
@Value("${spring.rabbitmq.username}")
private String USERNAME;
@Value("${spring.rabbitmq.password}")
private String PASSWORD;

@Bean
public CachingConnectionFactory connectionFactory() {
log.info("RabbitMQ链接信息：{}，{}，{},{}", HOST, PORT, USERNAME, PASSWORD);
CachingConnectionFactory connectionFactory = new CachingConnectionFactory(this.HOST, this.PORT);

connectionFactory.setUsername(this.USERNAME);
connectionFactory.setPassword(this.PASSWORD);
connectionFactory.setVirtualHost(this.VIRTUALHOST);
log.info("RabbitMQ连接成功");

return connectionFactory;

}

/**
* 死信队列交换机标识符属性值不能改，写死
*/
private static final String DEAD_LETTER_QUEUE_KEY = "x-dead-letter-exchange";
/**
* 死信队列交换机绑定键标识符属性值不能改，写死
*/
private static final String DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";

/**
* deadLetterExchange（direct类型交换机）
*
* @return
*/
@Bean("deadLetterExchange")
public Exchange deadLetterExchange() {
return ExchangeBuilder.directExchange("DEAD_LETTER_EXCHANGE").durable(true).build();
}

/**
* 声明一个死信队列
* x-dead-letter-exchange 对应死信交换机
* x-dead-letter-routing-key 对应死信队列
*/
@Bean("deadLetterQueue")
public Queue deadLetterQueue() {
//应该像个普通队列，里面多设置了两个参数，这个队列没有被消费或者超时则通过x-dead-letter-exchange 指明重新回到死信交换机 TEST_SIGN_EXCHANGE
//交换机
// 参数
Map<String, Object> args = new HashMap<>(2);
// 出现dead letter之后将dead letter重新发送到指定exchange
args.put(DEAD_LETTER_QUEUE_KEY, "DEAD_LETTER_EXCHANGE");
// 出现dead letter之后将dead letter重新按照指定的routing-key发送
args.put(DEAD_LETTER_ROUTING_KEY, "REDIRECT_KEY");
// name队列名字 durable是否持久化，true保证消息的不丢失, exclusive是否排他队列，如果一个队列被声明为排他队列，该队列仅对首次申明它的连接可见，并在连接断开时自动删除, autoDelete如果该队列没有任何订阅的消费者的话，该队列是否会被自动删除, arguments参数map
return new Queue("DEAD_LETTER_QUEUE", true, false, false, args);
}

/**
* 死信路由通过 DEAD_LETTER_KEY 绑定到死信队列上.
*/
@Bean
public Binding deadLetterBinding() {
return new Binding("DEAD_LETTER_QUEUE", Binding.DestinationType.QUEUE, "DEAD_LETTER_EXCHANGE", "DEAD_LETTER_KEY", null);

}

/**
* 死信路由通过 REDIRECT_KEY 绑定到转发队列上. 这个队列绑定的是当出现死信消息后重新转发给的队列
*/
@Bean
public Binding redirectBinding() {
return new Binding("REDIRECT_QUEUE", Binding.DestinationType.QUEUE, "DEAD_LETTER_EXCHANGE", "REDIRECT_KEY", null);
}
/**
* 定义死信队列转发队列. （和普通队列一样，这个队列是为了原有的消息没有被消费重新转发给一个新的队列）
*/
@Bean("redirectQueue")
public Queue redirectQueue() {
return new Queue("REDIRECT_QUEUE", true, false, false);
}

}

2.消息生产者 --》发送消息

import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

@Autowired
private RabbitTemplate rabbitTemplate;

/**
 * 延迟消息  ribbit MQ栗子
 *
 * @param messAge
 * @throws Exception
 */
@ApiOperation("死信发送消息")
@GetMapping("creatMessageDear")
public void creatMessageDear(@RequestParam("messAge") String messAge) throws Exception {
    //声明消息处理器 设置消息的编码以及消息的过期时间 时间毫秒值为字符串
    MessagePostProcessor messagePostProcessor = message -> {
        MessageProperties messageProperties = message.getMessageProperties();
        messageProperties.setMessageId(UUID.randomUUID().toString().replaceAll("-", ""));
        messageProperties.setContentEncoding("utf-8");
        //超时时间10秒    （我这里是延迟10秒，根据业务需要设置时间）
        messageProperties.setExpiration(String.valueOf(1000 * 10));
        return message;
    };
    Map<String, Object> dataMap = new HashMap<>();
    dataMap.put("msg", messAge);
    rabbitTemplate.convertAndSend("DEAD_LETTER_EXCHANGE", "DEAD_LETTER_KEY", dataMap, messagePostProcessor);
}

3.接受死信消息

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;



@Component
public class TestMQConsumer {

    /**
     * 监听转发队列  死信队列重新转发回这里
     *
     */
    @RabbitListener(queues = {"REDIRECT_QUEUE"})
    public void redirect(HashMap<String,Object> dataMap) throws IOException {
        System.out.println(dataMap.get("msg"));
        System.out.println("我是转发队列，这里执行逻辑业务");
    }
}

------------------------------

巴特西

Ribbit Mq 实现延迟消息

最新文章

热门文章