@Test public void testSendMessage() throws Throwable { TXContext ctx = new TXContextSupport(0L, 15099498831872L, "serial_86965", "purchase"); transactionMessageService.sendMessage(ctx, TransactionStatusEnum.CANCELLING); }
@Transactional(propagation = Propagation.REQUIRES_NEW) public void sendMessage(final TXContext ctx, TransactionStatusEnum txStatus) throws Throwable { TransactionInfo lockInfo = new TransactionInfo(); try { lockInfo.setParentId(ctx.getParentId()); lockInfo.setTxId(ctx.getTxId()); lockInfo = transactionRepository.lock(lockInfo).get(0); if (lockInfo == null) return; TransactionMessage message = new TransactionMessage(); message.setParentId(ctx.getTxId()); message.setBusinessId(ctx.getSerialNumber()); message.setBusinessType(ctx.getBusinessType()); message.setTxStatus(txStatus.getCode()); messageProducer.sendMessage(message); } catch (Exception e){ log.warn("Process 'handleMessage' failed, TXContext=" + ctx, e); TransactionInfo info = new TransactionInfo(); info.setTxId(ctx.getTxId()); info.setParentId(ctx.getParentId()); updateRetryCount(lockInfo); return; } TransactionInfo updInfo = new TransactionInfo(); updInfo.setParentId(ctx.getParentId()); updInfo.setTxId(ctx.getTxId()); updInfo.setTxStatus(TransactionStatusEnum.getNextStatus(txStatus).getCode()); transactionRepository.update(updInfo); log.info("Update Action TX status="+ TransactionStatusEnum.getMemo(updInfo.getTxStatus()) +", TXContext=" + ctx); }
TransactionMessageServiceImpl implements TransactionMessageService { @Transactional(propagation = Propagation.REQUIRES_NEW) public void sendMessage(final TXContext ctx, TransactionStatusEnum txStatus) throws Throwable { TransactionInfo lockInfo = new TransactionInfo(); try { lockInfo.setParentId(ctx.getParentId()); lockInfo.setTxId(ctx.getTxId()); lockInfo = transactionRepository.lock(lockInfo).get(0); if (lockInfo == null) return; TransactionMessage message = new TransactionMessage(); message.setParentId(ctx.getTxId()); message.setBusinessId(ctx.getSerialNumber()); message.setBusinessType(ctx.getBusinessType()); message.setTxStatus(txStatus.getCode()); messageProducer.sendMessage(message); } catch (Exception e){ log.warn("Process 'handleMessage' failed, TXContext=" + ctx, e); TransactionInfo info = new TransactionInfo(); info.setTxId(ctx.getTxId()); info.setParentId(ctx.getParentId()); updateRetryCount(lockInfo); return; } TransactionInfo updInfo = new TransactionInfo(); updInfo.setParentId(ctx.getParentId()); updInfo.setTxId(ctx.getTxId()); updInfo.setTxStatus(TransactionStatusEnum.getNextStatus(txStatus).getCode()); transactionRepository.update(updInfo); log.info("Update Action TX status="+ TransactionStatusEnum.getMemo(updInfo.getTxStatus()) +", TXContext=" + ctx); } }
TransactionMessageServiceImpl implements TransactionMessageService { @Transactional(propagation = Propagation.REQUIRES_NEW) public void sendMessage(final TXContext ctx, TransactionStatusEnum txStatus) throws Throwable { TransactionInfo lockInfo = new TransactionInfo(); try { lockInfo.setParentId(ctx.getParentId()); lockInfo.setTxId(ctx.getTxId()); lockInfo = transactionRepository.lock(lockInfo).get(0); if (lockInfo == null) return; TransactionMessage message = new TransactionMessage(); message.setParentId(ctx.getTxId()); message.setBusinessId(ctx.getSerialNumber()); message.setBusinessType(ctx.getBusinessType()); message.setTxStatus(txStatus.getCode()); messageProducer.sendMessage(message); } catch (Exception e){ log.warn("Process 'handleMessage' failed, TXContext=" + ctx, e); TransactionInfo info = new TransactionInfo(); info.setTxId(ctx.getTxId()); info.setParentId(ctx.getParentId()); updateRetryCount(lockInfo); return; } TransactionInfo updInfo = new TransactionInfo(); updInfo.setParentId(ctx.getParentId()); updInfo.setTxId(ctx.getTxId()); updInfo.setTxStatus(TransactionStatusEnum.getNextStatus(txStatus).getCode()); transactionRepository.update(updInfo); log.info("Update Action TX status="+ TransactionStatusEnum.getMemo(updInfo.getTxStatus()) +", TXContext=" + ctx); }  }
TransactionMessageServiceImpl implements TransactionMessageService { @Transactional(propagation = Propagation.REQUIRES_NEW) public void sendMessage(final TXContext ctx, TransactionStatusEnum txStatus) throws Throwable { TransactionInfo lockInfo = new TransactionInfo(); try { lockInfo.setParentId(ctx.getParentId()); lockInfo.setTxId(ctx.getTxId()); lockInfo = transactionRepository.lock(lockInfo).get(0); if (lockInfo == null) return; TransactionMessage message = new TransactionMessage(); message.setParentId(ctx.getTxId()); message.setBusinessId(ctx.getSerialNumber()); message.setBusinessType(ctx.getBusinessType()); message.setTxStatus(txStatus.getCode()); messageProducer.sendMessage(message); } catch (Exception e){ log.warn("Process 'handleMessage' failed, TXContext=" + ctx, e); TransactionInfo info = new TransactionInfo(); info.setTxId(ctx.getTxId()); info.setParentId(ctx.getParentId()); updateRetryCount(lockInfo); return; } TransactionInfo updInfo = new TransactionInfo(); updInfo.setParentId(ctx.getParentId()); updInfo.setTxId(ctx.getTxId()); updInfo.setTxStatus(TransactionStatusEnum.getNextStatus(txStatus).getCode()); transactionRepository.update(updInfo); log.info("Update Action TX status="+ TransactionStatusEnum.getMemo(updInfo.getTxStatus()) +", TXContext=" + ctx); }  @Transactional(propagation = Propagation.REQUIRES_NEW) void sendMessage(final TXContext ctx, TransactionStatusEnum txStatus); boolean isValidMessage(TransactionMessage message); void asyncHandleMessage(final TransactionMessage message); @Transactional // TODO 如果一个TX中同一个module参与多次，出现多条事务数据，此处事务需拆分处理 void handleMessage(TransactionMessage message); }
TransactionMessageServiceImpl implements TransactionMessageService { @Transactional(propagation = Propagation.REQUIRES_NEW) public void sendMessage(final TXContext ctx, TransactionStatusEnum txStatus) throws Throwable { TransactionInfo lockInfo = new TransactionInfo(); try { lockInfo.setParentId(ctx.getParentId()); lockInfo.setTxId(ctx.getTxId()); lockInfo = transactionRepository.lock(lockInfo).get(0); if (lockInfo == null) return; TransactionMessage message = new TransactionMessage(); message.setParentId(ctx.getTxId()); message.setBusinessId(ctx.getSerialNumber()); message.setBusinessType(ctx.getBusinessType()); message.setTxStatus(txStatus.getCode()); messageProducer.sendMessage(message); } catch (Exception e){ log.warn("Process 'handleMessage' failed, TXContext=" + ctx, e); TransactionInfo info = new TransactionInfo(); info.setTxId(ctx.getTxId()); info.setParentId(ctx.getParentId()); updateRetryCount(lockInfo); return; } TransactionInfo updInfo = new TransactionInfo(); updInfo.setParentId(ctx.getParentId()); updInfo.setTxId(ctx.getTxId()); updInfo.setTxStatus(TransactionStatusEnum.getNextStatus(txStatus).getCode()); transactionRepository.update(updInfo); log.info("Update Action TX status="+ TransactionStatusEnum.getMemo(updInfo.getTxStatus()) +", TXContext=" + ctx); }  @Transactional(propagation = Propagation.REQUIRES_NEW) void sendMessage(final TXContext ctx, TransactionStatusEnum txStatus); boolean isValidMessage(TransactionMessage message); void asyncHandleMessage(final TransactionMessage message); @Transactional // TODO 如果一个TX中同一个module参与多次，出现多条事务数据，此处事务需拆分处理 void handleMessage(TransactionMessage message);  }