4.1.5、消息幂等

@Component
public class FanoutProducer {

    @Autowired
    private AmqpTemplate amqpTemplate;

    public void send(String queueName) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("email", "xx@163.com");
        jsonObject.put("timestamp", System.currentTimeMillis());
        String jsonString = jsonObject.toJSONString();
        System.out.println("jsonString:" + jsonString);
        // 设置消息唯一id 保证每次重试消息id唯一
        Message message = MessageBuilder.withBody(jsonString.getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_JSON)
                .setContentEncoding("utf-8")
                .setMessageId(UUID.randomUUID() + "").build();
                //消息id设置在请求头里面 用UUID做全局ID
        amqpTemplate.convertAndSend(queueName, message);
    }
}

@RabbitListener(queues = "fanout_email_queue")
public void process(Message message) throws Exception {
     // 获取消息Id
     String messageId = message.getMessageProperties().getMessageId();  
     //id获取之
     if (messageId == null) {
            return;
     }
     String redisMsgId = ...//根据messageId去redis中查找消息
     if(未找到reidsMsgId) {
        String msg = new String(message.getBody(), "UTF-8");
              //消息内容获取之
         System.out.println("-----邮件消费者获取生产者消息-----------------" + "messageId:" + messageId + ",消息内容:" + msg);

         JSONObject jsonObject = JSONObject.parseObject(msg);
         // 获取email参数
         String email = jsonObject.getString("email");
         // 请求地址
         String emailUrl = "http://127.0.0.1:8083/sendEmail?email=" + email;
         JSONObject result = HttpClientUtils.httpGet(emailUrl);
         if (result == null) {
                 // 因为网络原因,造成无法访问,继续重试
                 throw new Exception("调用接口失败!");
         }
         System.out.println("执行结束....");
         //写入到redis中
     }

}