學習spring事務與消息隊列

更新時間：2016年10月12日 14:47:26 作者：登頂

這篇文章主要為大家詳細介紹了spring事務與消息隊列,具有一定的參考價值，感興趣的小伙伴們可以參考一下

在開發(fā)過程中，遇到一個bug，產(chǎn)生bug的原因是spring事務提交晚于消息隊列的生產(chǎn)消息，導致消息隊列消費消息時獲取到的數(shù)據(jù)不正確。這篇文章介紹問題的產(chǎn)生和一步步的解決過程。

一.問題的產(chǎn)生：

場景還原：接口中的一個方法，首先修改訂單狀態(tài)，然后向消息隊列中生產(chǎn)消息，消息隊列的消費者獲取到消息檢測訂單狀態(tài)，發(fā)現(xiàn)訂單狀態(tài)未更改。

代碼：

@Service(orderApi)
public class OrderApiImpl implements OrderApi {
  @Resource MqService mqService;
  @OrderDao orderDao;
   
  public void push(String orderId) {
    // 更新訂單狀態(tài)，之前的狀態(tài)是1
    updateStatus(orderId, 3);
    // 產(chǎn)生消息
    mqService.produce(orderId);
  }
  public viod updateStatus(String orderId, Integer status) {
    orderDao.updateStatus(orderId, status);
  }
}

問題產(chǎn)生原因：orderApi中的所有方法都有事務，事務類型PROPAGATION_REQUIRED，所以push方法對數(shù)據(jù)的操作會在push代碼全部執(zhí)行之后提交，而在事務提交之前消息隊列的消息已經(jīng)產(chǎn)生所以消息隊列中消費到的訂單從數(shù)據(jù)庫查詢出的狀態(tài)可能還為1。為了讓bug現(xiàn)象更明顯，可以在push方法最后添加：

try {
  Thread.sleep(10000);
} catch (InterruptedException e) {
  // TODO Auto-generated catch block
  e.printStackTrace();
}

這樣就會發(fā)現(xiàn)消費消息時，訂單狀態(tài)一定是未修改的。

二.問題的解決：

解決方案：在更新數(shù)據(jù)時，新建一個事物，保證更新代碼執(zhí)行完成后，更新數(shù)據(jù)庫的事務已被提交。（確保消息產(chǎn)生前數(shù)據(jù)庫操作已提交）

按照上述方案，我首先想到的是直接修改updateStatus方法的事務類型；我將此方法的事務類型改為PROPAGATION_REQUIRES_NEW（新建事務，如果當前存在事務，把當前事務掛起）。

但是這么做有兩點不合適：

　　1.強制修改了updateStaus的事務類型，可能影響其他流程。

　　2.未起到作用，updateStaus方法中沒有新建事務。

關(guān)于第二點的解釋：spring添加事務是通過BeanNameAutoProxyCreator實現(xiàn)的動態(tài)代理，只是給bean對象添加了事務，現(xiàn)在在類內(nèi)部調(diào)用方法，是不會觸發(fā)新事物的創(chuàng)建的。

所以在經(jīng)過以上嘗試后，我創(chuàng)建了一個新的類：

@Service("orderExtApi")
public class OrderExtApiImpl {
  @Resource OrderApi orderApi;
   
  public void updateStatusNewPropagation(String orderId) {
    orderApi.updateStatus(orderId);
  }
}

并為updateStatusNewPropagation方法添加事務PROPAGATION_REQUIRES_NEW

這個類就只是為了給orderApi中的updateStaus方法新起一個事務。

ok，到此為止bug已經(jīng)解決了。

但是代碼中還是存在問題：對數(shù)據(jù)庫的操作已經(jīng)提交，如果生產(chǎn)消息出現(xiàn)異常對業(yè)務邏輯來說還是錯誤的。所以需要檢測消息的產(chǎn)生是否完成。

最終orderApi中的代碼如下：

@Service(orderApi)
public class OrderApiImpl implements OrderApi {
  @Resource MqService mqService;
  @Resource OrderDao orderDao;
  @Resource OrderExtApiImpl orderExtApi;
   
  public void push(String orderId) {
    // 更新訂單狀態(tài)，之前的狀態(tài)是1
    orderExtApi.updateStatusNewPropagation(orderId, 3);
    // 產(chǎn)生消息--produce會檢測是否出現(xiàn)異常 當返回1時表示生產(chǎn)消息成功
    Response response = mqService.produce(orderId);
    if (response.getCode() != 1) {
      log.info("消息隊列生產(chǎn)消息異常：" + response.getErrorMsg())
      // 生產(chǎn)消息異常，重置狀態(tài) 等待下次重新執(zhí)行
      orderExtApi.updateStatusNewPropagation(orderId, 1);
    }
     
  }
  public viod updateStatus(String orderId, Integer status) {
    orderDao.updateStatus(orderId, status);
  }
}

以上就是本文的全部內(nèi)容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: