Find N3入網：最高支持16+1TB
OPPO將于近期登場的Find N3折疊屏目前已經正式入網，型號為PHN110。本次Find N3在外觀方面相比前兩代有很大的變化，不再是小號的橫向折疊屏，而是跟別的廠商一樣采用了較為常見的
本文給大家介紹一下在 Spring Boot 項目中如何集成消息隊列 RabbitMQ，包含對 RibbitMQ 的架構介紹、應用場景、坑點解析以及代碼實戰。最后文末有免費領取龍年紅包封面以及騰訊云社區答題領獎福利，歡迎大家領取。
我將使用 waynboot-mall 項目作為代碼講解，項目地址：https://github.com/wayn111/waynboot-mall。本文大綱如下，
圖片
圖片
RibbitMQ 是一個基于 AMQP 協議的開源消息隊列系統，具有高性能、高可用、高擴展等特點。通常作為在系統間傳遞消息的中間件，它可以實現異步處理、應用解耦、流量削峰等功能。
圖片
RibbitMQ 的主要組件介紹如下，
圖片
RabbitMQ 是一個非常強大和靈活的消息中間件，它可以應用于多種場景和需求。以下是一些常見的 RabbitMQ 應用場景和實戰經驗：
圖片
在使用 RabbitMQ 的過程中，有一些常見的問題需要注意：
在 waynboot-mall 項目中，消息層包含兩個模塊 waynboot-message-core 以及 waynboot-message-consumer，目錄結構如下，
|-- waynboot-message-core // 核心消息配置，供其他服務集成使用| |-- config| |-- constant| |-- dto|-- waynboot-message-consumer // 消息消費服務，訂閱隊列接收消息，調用其他服務執行一些具體的業務邏輯| |-- api| |-- config| |-- consumerwaynboot-message-core 包目錄說明如下，
waynboot-message-consumer 包目錄說明如下，
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> <version>${spring-boot.version}</version></dependency>
圖片
在 waynboot-mall 項目中，通過 yml 文件的 spring.rabbitmq.virtual-host=“/” 屬性來指定虛擬主機名稱。
建議大家在使用 RabbitMQ 時都配置好自己項目的虛擬主機名稱，來達到各系統資源隔離的目的。當然如果 RabbitMQ 服務只有一個項目在用，那就用默認的 / 作為虛擬主機名稱也是可以的。
小知識：出于多租戶和安全因素設計的，vhost 把 AMQP 的基本組件劃分到一個虛擬的分組中。每個 vhost 本質上就是一個 mini 版的 RabbitMQ 服務器，擁有自己的隊列、交換機、綁定和權限機制。當多個不同的用戶使用同一個 RabbitMQ 服務器時，可以劃分出多個虛擬主機。RabbitMQ 默認的虛擬主機路徑是 /。
在 waynboot-mall 項目中，用訂單消息來舉例，生產者發送消息需要經過三個步驟
public class MQConstants { public static final String ORDER_DIRECT_QUEUE = "order_direct_queue"; public static final String ORDER_DIRECT_EXCHANGE = "order_direct_exchange"; public static final String ORDER_DIRECT_ROUTING = "order_direct_routing";}@Configurationpublic class BusinessRabbitConfig { @Bean public Queue orderDirectQueue() { return new Queue(MQConstants.ORDER_DIRECT_QUEUE); } @Bean DirectExchange orderDirectExchange() { return new DirectExchange(MQConstants.ORDER_DIRECT_EXCHANGE); } @Bean Binding bindingOrderDirect() { return BindingBuilder.bind(orderDirectQueue()).to(orderDirectExchange()).with(MQConstants.ORDER_DIRECT_ROUTING); }}在 BusinessRabbitConfig 中，我們創建了訂單交換機、隊列以及路由綁定關系。在 Spring 項目中，項目啟動時，就會自動在 RabbitMQ 服務器上創建好這些東西。
交換機列表
隊列列表
生產者的消息發送確認主要包含兩部分，
producter -> rabbitmq broker exchange -> queue
waynboot-mall 項目的 yml 中關于 RabbitMQ 的相關配置如下，
spring: # 配置rabbitMq 服務器 rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest # 消息確認配置項 # 確認消息已發送到交換機(Exchange) publisher-confirm-type: correlated # 確認消息已發送到隊列(Queue) publisher-returns: true # 虛擬主機名稱 virtual-host: /可以看到，我們設置了 publisher-confirm-type 屬性為 correlated，表示開啟發布確認模式，用來確認消息已發送到交換機，publisher-confirm-type 有三個選項：
在 RabbitMQ 中，消息發送到交換機中也不代表消費者一定能接收到消息，所以我們還需要設置 publisher-returns 為 true 來表示確認交換機中消息已經發送到隊列里。true 表示開啟失敗回調，開啟后當消息無法路由到指定隊列時會觸發 ReturnCallback 回調。
接著是 RabbitTemplateConfig 的代碼，這里面會定義前面提到的 confirmCallBack、returnCallBack 相關代碼，
@Slf4j@Componentpublic class RabbitTemplateConfig { @Bean public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) { RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); // 設置開啟Mandatory，才能觸發回調函數,無論消息推送結果怎么樣都強制調用回調函數 rabbitTemplate.setMandatory(true); // 交換機收到消息回調 rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> log.info("消息發送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause)); // 隊列收到消息回調，如果失敗的話會進行 returnCallback 的回調處理，反之成功就不會回調。 rabbitTemplate.setReturnsCallback(returned -> { log.info("returnCallback: " + "消息：" + returned.getMessage()); log.info("returnCallback: " + "回應碼：" + returned.getReplyCode()); log.info("returnCallback: " + "回應信息：" + returned.getReplyText()); log.info("returnCallback: " + "交換機：" + returned.getExchange()); log.info("returnCallback: " + "路由鍵：" + returned.getRoutingKey()); }); return rabbitTemplate; }}在 RabbitTemplateConfig 類代碼里，我們可以設置 confirmCallBack、returnCallBack 回調函數后，監控生產者發送消息是否被交換機接收、以及交換機是否把消息發送到隊列中。
在 Spring Boot 項目中，集成了 spring-boot-starter-amqp 依賴后，就可以直接注入 RabbitTemplate 來發送消息。
這里用 waynboot-mall 項目中的異步下單流程舉例，代碼如下，
@Slf4j@Service@AllArgsConstructorpublic class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements IOrderService { private RabbitTemplate rabbitTemplate; @Override public R asyncSubmit(OrderVO orderVO) { OrderDTO orderDTO = new OrderDTO(); ... // 開始異步下單 String uid = IdUtil.getUid(); // 1. 創建消息ID，確認機制發送消息時，需要給每個消息設置一個全局唯一 id，以區分不同消息，避免 ack 沖突 CorrelationData correlationData = new CorrelationData(uid); // 2. 創建消息載體 Message ，AMQP 規范中定義的消息承載類，用來在生產者和消費者之前傳遞消息 Map<String, Object> map = new HashMap<>(); map.put("order", orderDTO); map.put("notifyUrl", WaynConfig.getMobileUrl() + "/callback/order/submit"); try { Message message = MessageBuilder .withBody(JSON.toJSONString(map).getBytes(Constants.UTF_ENCODING)) .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN) .setDeliveryMode(MessageDeliveryMode.PERSISTENT) .build(); // 3. 發送消息到 RabbitMQ 服務器，需要指定交換機、路由鍵、消息載體以及消息ID rabbitTemplate.convertAndSend(MQConstants.ORDER_DIRECT_EXCHANGE, MQConstants.ORDER_DIRECT_ROUTING, message, correlationData); } catch (UnsupportedEncodingException e) { log.error(e.getMessage(), e); } return R.success().add("actualPrice", actualPrice).add("orderSn", orderSn); }}waynboot-mall 項目中在使用 rabbitTemplate 發送消息時，按照如下步驟，大家可以參考
以上就是生產者發送消息時所有相關代碼了，接著我們看下消費者處理消息的相關代碼。
在 waynboot-mall 項目中，還是用訂單消息來舉例，消費者 yml 配置如下，
在 RabbitMQ 的消息消費環節，需要注意的一點就是，如果需要確保消費者不出現漏消費，則需要開啟消費者的手動 ack 模式。
spring: rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest ... listener: simple: # 消息確認方式，其有三種配置方式，分別是none、manual(手動ack) 和auto(自動ack) 默認auto acknowledge-mode: manual # 一個消費者最多可處理的nack（未確認）消息數量，默認是250 prefetch: 250 # 設置消費者數量 concurrency: 1在 yml 文件的消費者配置中，acknowledge-mode 屬性用于指定消息確認模式，有三種模式：
如果消費者在消費的過程中沒有拋出異常，則自動確認。
當消費者消費的過程中拋出 AmqpRejectAndDontRequeueException 異常的時候，則消息會被拒絕，且該消息不會重回隊列。
當拋出 ImmediateAcknowledgeAmqpException 異常，消息會被確認。
如果拋出其他的異常，則消息會被拒絕，但是與前兩個不同的是，該消息會重回隊列，如果此時只有一個消費者監聽該隊列，那么該消息重回隊列后又會推送給該消費者，會造成死循環的情況。
消費者配置中，prefetch 屬性用于指定消費者每次從隊列獲取的消息數量。
每個 customer 會在 MQ 預取一些消息放入內存的 LinkedBlockingQueue 中進行消費，這個值越高，消息傳遞的越快，但非順序處理消息的風險更高。如果 ack 模式為 none，則忽略。
prefetch 默認值以前是 1，這可能會導致高效使用者的利用率不足。從 spring-amqp 2.0 版開始，默認的 prefetch 值是 250，這將使消費者在大多數常見場景中保持忙碌，從而提高吞吐量。
不過在有些情況下，尤其是處理速度比較慢的大消息，消息可能在內存中大量堆積，消耗大量內存；以及對于一些嚴格要求順序的消息，prefetch 的值應當設置為 1。
對于低容量消息和多個消費者的情況（也包括單 listener 容器的 concurrency 配置）希望在多個使用者之間實現更均勻的消息分布，建議在手動 ack 下并設置 prefetch=1。
如果要保證消息的可靠不丟失，當 prefetch 大于 1 時，可能會出現因為服務宕機引起的數據丟失，故建議將 prefetch=1。
消費者配置中，concurrency 屬性設置的是對每個 listener 在初始化的時候設置的并發消費者的個數。在上面的 yml 配置中，cnotallow=1，即每個 Listener 容器將開啟一個線程去處理消息。在 2.0 以后的版本中，可以在注解中配置該參數，實例代碼如下，
@RabbitListener(queues = MQConstants.ORDER_DIRECT_QUEUE, concurrency = "2")public void process(Channel channel, Message message) throws IOException { String body = new String(message.getBody()); log.info("OrderPayConsumer 消費者收到消息: {}", body); ...}在 waynboot-mall 項目中，消費者監聽隊列代碼如下，
@Slf4j@Componentpublic class OrderPayConsumer { @Resource private RedisCache redisCache; @Resource private MobileApi mobileApi; @RabbitListener(queues = MQConstants.ORDER_DIRECT_QUEUE) public void process(Channel channel, Message message) throws IOException { // 1. 轉換訂單消息 String body = new String(message.getBody()); log.info("OrderPayConsumer 消費者收到消息: {}", body); // 2. 獲取消息ID String msgId = message.getMessageProperties().getHeader("spring_returned_message_correlation"); // 3. 獲取發送tag long deliveryTag = message.getMessageProperties().getDeliveryTag(); // 4. 消費消息冪等性處理 if (redisCache.getCacheObject(ORDER_CONSUMER_MAP.getKey()) != null) { // redis中包含該 key，說明該消息已經被消費過 log.error("msgId: {}，消息已經被消費", msgId); channel.basicAck(deliveryTag, false);// 確認消息已消費 return; } try { // 5. 下單處理 mobileApi.submitOrder(body); // 6. 手動ack，消息成功確認 channel.basicAck(deliveryTag, false); // 7. 設置消息已被消費標識 redisCache.setCacheObject(ORDER_CONSUMER_MAP.getKey(), msgId, ORDER_CONSUMER_MAP.getExpireSecond()); } catch (Exception e) { channel.basicNack(deliveryTag, false, false); log.error(e.getMessage(), e); } }}waynboot-mall 項目中在使用 RabbitListener 注解消費消息時，按照如下步驟，大家可以參考
其中參數 long deliveryTag 為消息的唯一序號也就是第三步獲取的發送 tag，第二個 boolean multiple 參數表示是否一次消費多條消息，false 表示只確認該序列號對應的消息，true 則表示確認該序列號對應的消息以及比該序列號小的所有消息，比如我先發送 2 條消息，他們的序列號分別為 2,3，并且他們都沒有被確認，還留在隊列中，那么如果當前消息序列號為 4，那么當 multiple 為 true，則序列號為 2、3 的消息也會被一同確認。
這篇文章給大家講解了在 Spring Boot 項目中如何集成消息隊列 RabbitMQ 用于業務邏輯解耦，有架構介紹、應用場景、坑點解析、代碼實戰 4 個部分，能帶領大家比較全面的了解一波 RabbitMQ。大家在自己的項目中如果需要引入 RabbitMQ 時，都可以參考本文的代碼實戰配置，幫助大家快速集成、避免踩坑。
本文鏈接：http://www.www897cc.com/showinfo-26-70467-0.htmlSpring Boot項目集成RabbitMQ實戰以及坑點講解
聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。郵件：2376512515@qq.com
上一篇： 利用Nacos實現Seata事務模式（XA與AT）的快速配置與靈活切換
下一篇： 字節碼增強技術，不止有 Java Proxy、 Cglib 和 Javassist 還有 Byte Buddy
Copyright ? 2016-2023 天津谷騏科技有限公司 版權所有 sitemap.xml
違法及侵權請聯系：2376512515@qq.com 津ICP備18001702號
津公網安備 12010102000574號