java分布式事务处理 http://my.oschina.net/u/1054538/blog/718343
在大型系统架构时我们会进行分库设计，比如用户库、订单库。如果采用了dubbo会产生服务,如果目前有两个服务，用户服务和订单服务。
实际业务中，用户下单支付成功后，并改变用户的状态或增加用户的积分。这样过程中就会产生事务问题。这里我们采用最终事务一致性。
大致实现思路，把分布式事务切割成小事务，用消息队列消除分布式事务。实现方式如下：
订单功能的小事务如下：
首先：订单服务。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 |
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; } }); |
用户服务监听消息队列
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 |
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); } }); } |
消息队列配置
|
1 |
<amq:redeliveryPolicy id="activeMQRedeliveryPolicy" destination="#defaultDestination" redeliveryDelay="1000" maximumRedeliveries="10" /> |
消息队列重试次数10次，如果一次失败，可以多试几次。
如果消息队列最终失败，则监听失败信息，采用事务补偿机制，删除之前增加的订单或其他处理。
