前言

再多的話就不說了，這個是接著上一講：
【一起學設計模式】狀態模式+裝飾器模式+簡單工廠模式實戰：（一）提交個訂單我到底經歷了什麼鬼？
一起的，一些多餘的贅述請先看這個篇文章。

業務場景

一圖流，還是上一篇文章中一樣的圖，接下來我們就梳理下總結模式、觀察者模式、備忘錄模式的應用：

訂單中心：
1、訂單中心建立訂單
2、訂單狀態流轉（狀態模式）
3、記錄操作日誌（裝飾器模式+簡單工廠模式）
4、訂單中心通知 庫存中心更新庫存

排程中心：
1、庫存中心更新本地庫存（使用命令模式+模板方法模式+工廠模式）
這個上講已經說過：[【一起學設計模式】命令模式+模板方法+工廠方法實戰: 如何優雅的更新商品庫存...][5]

2、將更新庫存資料放到訊息中，排程中心消費訊息（中介模式）
3、放入訊息佇列中，判斷佇列是否放滿了，如果放滿了需要建立離線儲存（備忘錄模式）
4、非同步監聽訊息處理結果（觀察者模式）

這個模型應該很簡單，我們來一步步拆解 一步步程式碼分析

庫存中傳送訊息給庫存中心

下單後，訂單中心呼叫庫存中心 扣減庫存，然後庫存中心呼叫排程中心，排程中心再去自身庫存扣減、WMS揀貨單生成、發貨單生成等一些列排程任務。

為了解耦，庫存中心將需要傳送的內容放到一個記憶體佇列中，排程中心非同步去消費訊息。

程式碼實現

1. 庫存中心提供給訂單中心介面

   /**
    * 通知庫存中心，“提交訂單”事件發生了
    * @param orderDTO 訂單DTO
    * @return 處理結果
    */
   @Override
   public Boolean informSubmitOrderEvent(OrderInfoDTO orderDTO) {
       try {
           // 更新本地庫存
           // do logic
   
           // 傳送非同步訊息到記憶體佇列
           StockUpdateMessage message = new StockUpdateMessage();
           message.setId(UUID.randomUUID().toString().replace("-", ""));   
           message.setOperation(GoodsStockUpdateOperation.SUBMIT_ORDER);
           message.setParameter(orderDTO);  
           goodsStockUpdateQueue.put(message);
   
           // 監聽非同步處理結果
           goodsStockUpdateManager.observe(message.getId());  
       } catch (Exception e) {
           logger.error("error", e); 
           return false;
       }
       return true;
   }
    
   

2. 自定義一個記憶體佇列

   
   /**
    * 商品庫存更新訊息的佇列介面
    * @author wangmeng
    *
    */
   public interface StockUpdateQueue {
   
       /**
        * 將一個訊息放入佇列
        * @param message 訊息
        * @throws Exception
        */
       void put(StockUpdateMessage message) throws Exception;
   
       /**
        * 直接將訊息放入佇列
        * @param message
        * @throws Exception
        */
       void putDirect(StockUpdateMessage message) throws Exception;
   
       /**
        * 從佇列中取出一個訊息
        * @return
        * @throws Exception
        */
       StockUpdateMessage take() throws Exception;
   
       /**
        * 獲取佇列大小
        * @return
        * @throws Exception
        */
       Integer size() throws Exception;
   }
   
   /**
    * 商品庫存更新佇列實現類
    * @author wangmeng
    *
    */
   @Component
   public class StockUpdateQueueImpl implements StockUpdateQueue {
   
       private static final Integer QUEUE_MAX_SIZE = 1000;
   
       /**
        * 離線儲存管理元件
        */
       @Autowired
       private OfflineStorageManager offlineStorageManager;
   
       /**
        * 商品庫存更新佇列
        */
       private ArrayBlockingQueue<StockUpdateMessage> queue = 
               new ArrayBlockingQueue<StockUpdateMessage>(QUEUE_MAX_SIZE); 
   
       /**
        * 將一個訊息放入佇列
        * @param message 訊息
        * @throws Exception
        */
       @Override
       public void put(StockUpdateMessage message) throws Exception {
           queue.put(message); 
       }
   
       /**
        * 從佇列中取出一個訊息
        * @return
        * @throws Exception
        */
       @Override
       public StockUpdateMessage take() throws Exception { 
           return queue.take();
       }
   
       /**
        * 直接將訊息放入佇列
        * @param message
        * @throws Exception
        */
       @Override
       public void putDirect(StockUpdateMessage message) throws Exception {
           queue.put(message); 
       }
   
       /**
        * 獲取佇列大小
        * @return
        * @throws Exception
        */
       @Override
       public Integer size() throws Exception {
           return queue.size();
       }
   
   }
    
   

3. 自定義訊息體

   /**
    * 商品庫存更新訊息
    * @author wangmeng
    *
    */
    @Data
   public class StockUpdateMessage {
   
       /**
        * id
        */
       private String id;
       /**
        * 商品庫存更新操作
        */
       private Integer operation;
       /**
        * 核心引數資料
        */
       private Object parameter;
   }

4. 排程中心訊息消費者

   /**
    * 庫存更新訊息消費者
    * @author wangmeng
    *
    */
   @Component
   public class ScheduleStockUpdateMessageConsumer extends Thread {
   
       private static final Logger logger = LoggerFactory.getLogger(
               ScheduleStockUpdateMessageConsumer.class);
   
       /**
        * 庫存更新訊息佇列
        */
       @Autowired
       private StockUpdateQueue stockUpdateQueue;
       /**
        * 排程中心介面
        */
       @Autowired
       private ScheduleService scheduleService;
       /**
        * 庫存中心的訊息管理器
        */
       @Autowired
       private StockUpdateResultManager stockUpdateResultManager;
   
       /**
        * 消費庫存更新訊息
        */
       @Override
       public void run() {
           while(true) {
               try {
                   StockUpdateMessage message = stockUpdateQueue.take();
                   if(!isOrderRelatedMessage(message)) {
                       continue;
                   }
                   OrderInfoDTO order = getOrderFromMessage(message);
                   processMessage(message, order);
                   stockUpdateResultManager.inform(message.getId(), true); 
               } catch (Exception e) {
                   logger.error("error", e); 
               }
           }
       }
   
       /**
        * 是否是訂單相關的操作
        * @param message 訊息
        * @return 是否是訂單相關的操作
        * @throws Exception
        */
       private Boolean isOrderRelatedMessage(StockUpdateMessage message) throws Exception {
           return GoodsStockUpdateOperation.SUBMIT_ORDER.equals(message.getOperation())
                   || GoodsStockUpdateOperation.CANCEL_ORDER.equals(message.getOperation()) 
                   || GoodsStockUpdateOperation.PAY_ORDER.equals(message.getOperation());
       }
   
       /**
        * 從訊息中獲取訂單
        * @param message 訊息
        * @return 訂單
        * @throws Exception
        */
       private OrderInfoDTO getOrderFromMessage(StockUpdateMessage message) throws Exception {
           return (OrderInfoDTO) message.getParameter();
       }
   
       /**
        * 處理訊息
        * @param order 訂單 
        * @return 處理結果
        * @throws Exception
        */
       private Boolean processMessage(StockUpdateMessage message, 
               OrderInfoDTO order) throws Exception {
           if(GoodsStockUpdateOperation.SUBMIT_ORDER.equals(message.getOperation())) {
               return scheduleService.informSubmitOrderEvent(order);
           } else if(GoodsStockUpdateOperation.CANCEL_ORDER.equals(message.getOperation())) {
               return scheduleService.informCancelOrderEvent(order);
           } else if(GoodsStockUpdateOperation.PAY_ORDER.equals(message.getOperation())) {
               return scheduleService.informPayOrderEvent(order);
           }
           return false;
       }
   
   }

監聽訊息佇列，防止訊息滿載

這裡我們用的是一個記憶體阻塞佇列，那麼我們就需要考慮如果消費者出現異常或者消費過慢的情況導致訊息阻塞該怎麼辦？

這裡我們使用備忘錄模式 記錄佇列中佇列是否滿載，如果是則加入到離線儲存，儲存到db中。如果佇列恢復size=0 再將離線資料放入佇列中。

程式碼實現

1. 訊息放入佇列

   /**
    * 將一個訊息放入佇列
    * @param message 訊息
    * @throws Exception
    */
   public void put(StockUpdateMessage message) throws Exception {
       // 每次要往記憶體佇列放訊息之前，先檢查一下離線儲存標識
       // 如果觸發了離線儲存，直接就往離線儲存去寫入，不要走後面的邏輯了
       // 寫完離線儲存之後，需要檢查一下記憶體佇列的大小，如果記憶體佇列已經清零，則啟動一個後臺執行緒
       // 讓後臺執行緒去將離線儲存中的資料恢復寫入記憶體佇列中
       if(offlineStorageManager.getOffline()) {
           offlineStorageManager.store(message); 
   
           if(queue.size() == 0) {
               new OfflineResumeThread(offlineStorageManager, this).start(); 
           }
   
           return;
       }
   
       // 如果記憶體佇列已經滿了，此時就觸發離線儲存
       if(QUEUE_MAX_SIZE.equals(queue.size())) {  
           offlineStorageManager.store(message); 
           offlineStorageManager.setOffline(true);
           return;
       }
   
       queue.put(message); 
   }

2. 離線儲存管理器

   /**
    * 離線儲存管理元件介面
    * @author wangmeng
    *
    */
   public interface OfflineStorageManager {
   
       /**
        * 離線儲存庫存更新訊息
        * @param message 庫存更新訊息
        * @throws Exception
        */
       void store(StockUpdateMessage message) throws Exception;
   
       /**
        * 獲取離線儲存標識
        * @return 離線儲存標識
        * @throws Exception
        */
       Boolean getOffline() throws Exception;
   
       /**
        * 設定離線儲存標識
        * @param offline 離線儲存標識
        * @throws Exception
        */
       void setOffline(Boolean offline) throws Exception;
   
       /**
        * 所謂的迭代器模式，什麼時候用？
        * 
        * 其實只有一個場景，就是如果你需要基於一些不支援迭代的資料，來讓我們業務程式碼進行迭代
        * 那麼你自己就要去實現基於那個資料的一套迭代程式碼
        * 以迭代器的方式返回回去給業務方，來通過你定義的迭代器，進行資料的迭代
        * 
        * mysql資料庫，本身是不支援迭代式訪問的，但是我們可以自己實現一套基於mysql的迭代訪問的程式碼
        * 把一個迭代器給返回回去
        * 
        * 比如有的時候，我們可能還需要基於es、redis的資料，來提供業務方迭代式訪問的功能，那麼此時就只能我們自己
        * 去封裝迭代器，在裡面封裝基於es、redis的迭代訪問資料的邏輯
        * 
        */
       /**
        * 獲取迭代器
        * @return 迭代器
        * @throws Exception
        */
       OfflineStorageIterator iterator() throws Exception;
   
       /**
        * 批量刪除庫存更新訊息
        * @param stockUpdateMessages 庫存更新訊息
        * @throws Exception
        */
       void removeByBatch(List<StockUpdateMessage> stockUpdateMessages) throws Exception;
   }
   
   /**
    * 離線儲存管理元件
    * @author wangmeng
    *
    */
   @Component
   public class OfflineStorageManagerImpl implements OfflineStorageManager {
   
       /**
        * 庫存更新訊息管理模組DAO元件
        */
       @Autowired
       private StockUpdateMessageDAO stockUpdateMessageDAO;
   
       /**
        * 是否觸發離線儲存的標識
        */
       private Boolean offline = false;
   
       /**
        * 離線儲存庫存更新訊息
        * @param message 庫存更新訊息
        * @throws Exception
        */
       @Override
       public void store(StockUpdateMessage message) throws Exception {
           StockUpdateMessageDO stockUpdateMessageDO = createStockUpdateMessageDO(message);
           stockUpdateMessageDAO.save(stockUpdateMessageDO);
       }
   
       /**
        * 建立庫存更新訊息DO物件
        * @param message 庫存更新訊息
        * @return 庫存更新訊息DO物件
        * @throws Exception
        */
       private StockUpdateMessageDO createStockUpdateMessageDO(
               StockUpdateMessage message) throws Exception {
           StockUpdateMessageDO stockUpdateMessageDO = new StockUpdateMessageDO();
           stockUpdateMessageDO.setMessageId(message.getId());
           stockUpdateMessageDO.setOperation(message.getOperation());
           stockUpdateMessageDO.setParameter(JSONObject.toJSONString(message.getParameter())); 
           stockUpdateMessageDO.setParamterClazz(message.getParameter().getClass().getName());  
           stockUpdateMessageDO.setGmtCreate(new Date()); 
           stockUpdateMessageDO.setGmtModified(new Date()); 
           return stockUpdateMessageDO;
       }
   
       /**
        * 獲取離線儲存標識
        * @return 離線儲存標識
        * @throws Exception
        */
       @Override
       public Boolean getOffline() throws Exception {
           return offline;
       }
   
       /**
        * 設定離線儲存標識
        * @param offline 離線儲存標識
        * @throws Exception
        */
       @Override
       public void setOffline(Boolean offline) throws Exception {
           this.offline = offline;
       }
   
       /**
        * 批量刪除庫存更新訊息
        * @param stockUpdateMessages 庫存更新訊息
        * @throws Exception
        */
       @Override
       public void removeByBatch(List<StockUpdateMessage> stockUpdateMessages) throws Exception {
           StringBuilder builder = new StringBuilder("");
           for(int i = 0; i < stockUpdateMessages.size(); i++) {
               builder.append(stockUpdateMessages.get(i).getId());
               if(i < stockUpdateMessages.size() - 1) {
                   builder.append(","); 
               }
           }
           stockUpdateMessageDAO.removeByBatch(builder.toString());
       }
   
       /**
        * 獲取離線資料迭代器
        * @throws Exception
        */
       @Override
       public OfflineStorageIterator iterator() throws Exception {
           return new OfflineStorageIteratorImpl();
       }
   
       /**
        * 離線資料迭代器
        * @author zhonghuashishan
        *
        */
       public class OfflineStorageIteratorImpl implements OfflineStorageIterator {
   
           /**
            * 判斷是否還有下一批庫存更新訊息
            * @return 是否還有下一批庫存更新訊息
            * @throws Exception
            */
           @Override
           public Boolean hasNext() throws Exception {
               return stockUpdateMessageDAO.count().equals(0L) ? false : true;
           }
   
           /**
            * 獲取下一批庫存更新訊息
            * @return 下一批庫存更新訊息
            * @throws Exception
            */
           @Override
           public List<StockUpdateMessage> next() throws Exception {
               List<StockUpdateMessage> stockUpdateMessages = new ArrayList<StockUpdateMessage>();
   
               List<StockUpdateMessageDO> stockUpdateMessageDOs = 
                       stockUpdateMessageDAO.listByBatch();
               for(StockUpdateMessageDO stockUpdateMessageDO : stockUpdateMessageDOs) {
                   StockUpdateMessage stockUpdateMessage = new StockUpdateMessage();
                   stockUpdateMessage.setId(stockUpdateMessageDO.getMessageId()); 
                   stockUpdateMessage.setOperation(stockUpdateMessageDO.getOperation()); 
                   stockUpdateMessage.setParameter(JSONObject.parseObject(stockUpdateMessageDO.getParameter(), 
                           Class.forName(stockUpdateMessageDO.getParamterClazz())));  
                   stockUpdateMessages.add(stockUpdateMessage);
               }
   
               return stockUpdateMessages;
           }
       }
   }

3. 離線資料恢復類

   /**
    * 離線資料恢復執行緒
    * @author wangmeng
    *
    */
   public class OfflineResumeThread extends Thread {
   
       private static final Logger logger = LoggerFactory.getLogger(OfflineResumeThread.class);
   
       /**
        * 離線儲存管理元件
        */
       private OfflineStorageManager offlineStorageManager;
       /**
        * 庫存更新佇列
        */
       private StockUpdateQueue stockUpdateQueue;
   
       /**
        * 建構函式
        * @param offlineStorageManager 離線儲存管理元件
        */
       public OfflineResumeThread(OfflineStorageManager offlineStorageManager,
               StockUpdateQueue stockUpdateQueue) {
           this.offlineStorageManager = offlineStorageManager;
           this.stockUpdateQueue = stockUpdateQueue;
       }
   
       /**
        * 執行執行緒
        */
       @Override
       public void run() {
           try {
               // 如果表中還有資料的話
               OfflineStorageIterator offlineStorageIterator = offlineStorageManager.iterator();
   
               while(offlineStorageIterator.hasNext()) {
                   try {
                       // 每次就從mysql中查詢50條資料，批量查詢，批量處理，批量刪除
                       List<StockUpdateMessage> stockUpdateMessages = offlineStorageIterator.next();
   
                       // 將這批資料寫入記憶體佇列中
                       for(StockUpdateMessage message : stockUpdateMessages) {
                           stockUpdateQueue.putDirect(message);
                       }
   
                       // 批量刪除這批資料
                       offlineStorageManager.removeByBatch(stockUpdateMessages); 
                   } catch (Exception e) {
                       logger.error("error", e); 
                   }
               }
   
               // 此時mysql中的資料全部恢復完，更新記憶體標識
               offlineStorageManager.setOffline(false); 
           } catch (Exception e) {
               logger.error("error", e); 
           }
       }
   }

庫存中心非同步監聽訊息消費結果

我們在上面 其實已經有了一端程式碼 是描述非同步監聽消費結果的，這裡再來具體貼下 觀察者、被觀察者的程式碼。

程式碼實現

1. 被觀察者

   /**
    * 商品庫存更新結果觀察目標
    * @author wangmeng
    *
    */
   public class StockUpdateObservable extends Observable {
   
       /**
        * 訊息id
        */
       private String messageId;
   
       /**
        * 建構函式
        * @param messageId 訊息id
        */
       public StockUpdateObservable(String messageId) {
           this.messageId = messageId;
       }
   
       /**
        * 設定商品庫存更新結果
        * @param result 商品庫存更新結果
        */
       public void setResult(Boolean result) {
           StockUpdateResult goodsStockUpdateResult = new StockUpdateResult();
           goodsStockUpdateResult.setMessageId(messageId); 
           goodsStockUpdateResult.setResult(result); 
   
           this.setChanged();
           this.notifyObservers(goodsStockUpdateResult);  
       }
   
       public String getMessageId() {
           return messageId;
       }
   }

2. 觀察者

   /**
    * 商品庫存更新結果觀察者
    * @author wangmeng
    *
    */
   @Component
   public class StockUpdateObserver implements Observer {
   
       private static final Logger logger = LoggerFactory.getLogger(
               StockUpdateObserver.class);
   
       /**
        * 通知非同步處理結果
        */
       @Override
       public void update(Observable o, Object arg) {
           StockUpdateResult result = (StockUpdateResult) arg;
           logger.info("商品庫存更新訊息[messageId=" + result.getMessageId() + "]" 
                   + "的非同步處理結果為：" + result.getResult());    
       }
   }

3. 新增觀察者

observe方法是訂單中心通知庫存中心更新庫存的時候呼叫的，庫存中心給排程中心傳送非同步訊息，然後將這個訊息的messageId加入到觀察者中。

inform方法是排程中心的訊息消費者呼叫的，如果消費成功，排程中心會呼叫inform方法，設定result=true

```java
/**
 * 商品庫存更新結果管理元件
 * @author wangmeng
 *
 */
@Component
public class StockUpdateResultManagerImpl 
        implements StockUpdateResultManager {

    /**
     * 商品庫存更新結果map
     */
    private Map<String, StockUpdateObservable> observableMap = 
            new ConcurrentHashMap<String, StockUpdateObservable>();

    /**
     * 商品庫存更新結果觀察者
     */
    @Autowired
    private StockUpdateObserver observer;

    /**
     * 設定對商品庫存更新結果的觀察
     * @param messageId 訊息id
     * @param result 商品庫存更新結果
     * @param observer 商品庫存更新結果的觀察者
     */
    @Override
    public void observe(String messageId) {
        StockUpdateObservable observable = new StockUpdateObservable(messageId);
        observable.addObserver(observer);  
        observableMap.put(messageId, observable);
    }

    /**
     * 獲取商品庫存更新結果的觀察目標
     * @param messageId 商品庫存更新訊息id
     * @return 商品庫存更新結果的觀察目標
     */
    @Override
    public void inform(String messageId, Boolean result) {
        StockUpdateObservable observable = observableMap.get(messageId);
        observable.setResult(result); 
        observableMap.remove(messageId);
    }

    /**
     * 獲取庫存更新結果觀察目標
     * @param messageId 訊息id
     * @return
     */
    @Override
    public StockUpdateObservable getObservable(String messageId) {
        return observableMap.get(messageId);
    }

}
```

總結

本篇內容有點多，主要是分為了三大塊，然後結合了中介者模式、備忘錄模式、觀察者模式。

其中在離線訊息恢復的類中還是用了迭代器模式。

程式碼做了簡單的抽離，我相信讀起來還是很輕鬆的，設計模式系列要先告一段落了，這幾篇文章涉及了 一些常用的設計模式，後面如果有新的模式還會繼續連載更新。

申明

本文章首發自本人部落格：https://www.cnblogs.com/wang-meng 和公眾號：壹枝花算不算浪漫，如若轉載請標明來源!

感興趣的小夥伴可關注個人公眾號：壹枝花算不算浪漫

相關推薦