jms+jpa

配置

    @Bean
    public ConnectionFactory connectionFactory(){
        ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://localhost:61616");
        //保证最终一致性
        TransactionAwareConnectionFactoryProxy proxy = new TransactionAwareConnectionFactoryProxy();
        proxy.setTargetConnectionFactory(cf);
        proxy.setSynchedLocalTransactionAllowed(true);
        return proxy;
    }

    @Bean
    public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory){
        JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
        jmsTemplate.setSessionTransacted(true);
        return jmsTemplate;
    }

    @Autowired
    CustomerRepository customerRepository;

    @Autowired
    JmsTemplate jmsTemplate;

    @Transactional
    @JmsListener(destination = "customer:msg:new")
    public void handle(String msg){
        log.info("get msg1:{}",msg);
        Customer customer = new Customer();
        customer.setUserName(msg);
        customer.setDeposit(100);
        customerRepository.save(customer);
        if(msg.contains("error1")){
            throw new RuntimeException("Error1");
        }

        jmsTemplate.convertAndSend("customer:msg:reply",msg );
        if(msg.contains("error2")){
            throw new RuntimeException("Error2");
        }
    }

    @Transactional
    public Customer create(Customer customer){
        log.info("CustomerService create customer:{}",customer.getUserName());
        customer = customerRepository.save(customer);
        if(customer.getUserName().contains("error1")){
            throw new RuntimeException("Error1");
        }
        jmsTemplate.convertAndSend("customer:msg:reply",customer.getUserName());
        if(customer.getUserName().contains("error2")){
            throw new RuntimeException("Error2");
        }
        return customer;
    }