spring:
  rabbitmq:
    # 連接地址
    host: 127.0.0.1
    # 端口號
    port: 5672
    # 賬號
    username: guest
    # 密碼
    password: guest
    # 地址(類似于數(shù)據(jù)庫的概念)
    virtual-host: /admin_vhost
    # 消費者監(jiān)聽相關配置
    listener:
      simple:
        retry:
          # 開啟消費者(程序出現(xiàn)異常)重試機制，默認開啟并一直重試
          enabled: true
          # 最大重試次數(shù)
          max-attempts: 5
          # 重試間隔時間(毫秒)
          initial-interval: 3000

@Component
public class FanoutProducer {
 @Autowired
 private AmqpTemplate amqpTemplate;

 public void send(String queueName) {
  String msg = "my_fanout_msg:" + System.currentTimeMillis();
  //請求頭設置消息id（messageId）
  Message message = MessageBuilder.withBody(msg.getBytes()).setContentType(MessageProperties.CONTENT_TYPE_JSON)
    .setContentEncoding("utf-8").setMessageId(UUID.randomUUID() + "").build();
  System.out.println(msg + ":" + msg);
  amqpTemplate.convertAndSend(queueName, message);
 }
}

@RabbitListener(queues = "fanout_email_queue")
 public void process(Message message) throws Exception {
  // 獲取消息Id
  String messageId = message.getMessageProperties().getMessageId();
  String msg = new String(message.getBody(), "UTF-8");
  //② 判斷唯一Id是否被消費，消息消費成功后將id和狀態(tài)保存在日志表中，我們從（①步驟）表中獲取并判斷messageId的狀態(tài)即可
  //從redis中獲取messageId的value
  String value = redisUtils.get(messageId)+"";
  if(value.equals("1") ){ //表示已經消費
   return; //結束
  }
  System.out.println("郵件消費者獲取生產者消息" + "messageId:" + messageId + ",消息內容:" + msg);
  JSONObject jsonObject = JSONObject.parseObject(msg);
  // 獲取email參數(shù)
  String email = jsonObject.getString("email");
  // 請求地址
  String emailUrl = "http://127.0.0.1:8083/sendEmail?email=" + email;
  JSONObject result = HttpClientUtils.httpGet(emailUrl);
  if (result == null) {
   // 因為網絡原因,造成無法訪問,繼續(xù)重試
   throw new Exception("調用接口失敗!");
  }
  System.out.println("執(zhí)行結束....");
  //① 執(zhí)行到這里已經消費成功，我們可以修改messageId的狀態(tài)，并存入日志表(可以存到redis中，key為消息Id、value為狀態(tài))
 }

spring:
  rabbitmq:
    # 連接地址
    host: 127.0.0.1
    # 端口號
    port: 5672
    # 賬號
    username: guest
    # 密碼
    password: guest
    # 地址(類似于數(shù)據(jù)庫的概念)
    virtual-host: /admin_vhost
    # 消費者監(jiān)聽相關配置
    listener:
      simple:
        retry:
          # 開啟消費者(程序出現(xiàn)異常)重試機制，默認開啟并一直重試
          enabled: true
          # 最大重試次數(shù)
          max-attempts: 5
          # 重試間隔時間(毫秒)
          initial-interval: 3000
        # 開啟手動ack
        acknowledge-mode: manual

Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG); 手動ack
channel.basicAck(deliveryTag, false);手動簽收

//郵件隊列
@Component
public class FanoutEamilConsumer {
 @RabbitListener(queues = "fanout_email_queue")
 public void process(Message message, @Headers Map<String, Object> headers, Channel channel) throws Exception {
  System.out
    .println(Thread.currentThread().getName() + ",郵件消費者獲取生產者消息msg:" + new String(message.getBody(), "UTF-8")
      + ",messageId:" + message.getMessageProperties().getMessageId());
  // 手動ack
  Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
  // 手動簽收
  channel.basicAck(deliveryTag, false);
 }
}