java分佈式事務處理--最終事務一致性

在大型系統架構時咱們會進行分庫設計，好比用戶庫、訂單庫。若是採用了dubbo會產生服務,若是目前有兩個服務，用戶服務和訂單服務。

實際業務中，用戶下單支付成功後，並改變用戶的狀態或增長用戶的積分。這樣過程當中就會產生事務問題。這裏咱們採用最終事務一致性。

    大體實現思路，把分佈式事務切割成小事務，用消息隊列消除分佈式事務。實現方式以下：

訂單功能的小事務以下：

首先：訂單服務。

jmsTemplate.setSessionTransacted(true);
		transactionTemplate.execute(new TransactionCallback<String>()
		{

			@Override
			public String doInTransaction(TransactionStatus status)
			{
				// TODO Auto-generated method stub
				Connection connection = null;
				Session session = null;
				try
				{
					String orderId = System.currentTimeMillis() + "";
					String sql = "insert into order (order_id,user_id) values (?,?)";
					jdbcTemplate.update(sql, new Object[]
					{ orderId, userId });
					connection = jmsTemplate.getConnectionFactory().createConnection();
					session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
					Destination destination = session.createQueue("transactionQueue");
					MessageProducer producer = session.createProducer(destination);
					producer.setDeliveryMode(DeliveryMode.PERSISTENT);
					String text = "orderid";
					MapMessage message = session.createMapMessage();//Message(text);
					message.setString("order_id", orderId);
					message.setString("user_id", userId);
					message.setString("status", "1");
					producer.send(message);
					session.commit();
				} catch (Exception ex)
				{
					// TODO: handle exception
					status.setRollbackOnly();
					try
					{
						session.rollback();
					} catch (JMSException e)
					{
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
					ex.printStackTrace();
				} finally
				{
					try
					{
						session.close();
					} catch (JMSException e)
					{
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
				return null;
			}
		});

 

用戶服務監聽消息隊列

public void handlerMessage(final MapMessage mapMessage, final Session session) throws JMSException
	{
		getTransactionTemplate().execute(new TransactionCallback<String>()
		{

			@Override
			public String doInTransaction(TransactionStatus status)
			{
				int result = 0;
				try
				{
					//String status, String user_id, String order_id
					updateUserLevel(mapMessage.getString("status"), 
                                    mapMessage.getString("user_id"),
							        mapMessage.getString("order_id"));
				} catch (Exception e)
				{
					try
					{
						session.rollback();
					} catch (JMSException ex)
					{
						logger.error("JMS事務回滾異常", ex);
						ex.printStackTrace();
					}
					status.setRollbackOnly();

				}

				return String.valueOf(result);

			}
		});

	}

消息隊列配置

​
<amq:redeliveryPolicy id="activeMQRedeliveryPolicy" destination="#defaultDestination"  redeliveryDelay="1000" maximumRedeliveries="10" />

​

 

消息隊列重試次數10次，若是一次失敗，能夠多試幾回。

若是消息隊列最終失敗，則監聽失敗信息，採用事務補償機制，刪除以前增長的訂單或其餘處理。