业务场景：

在一个加了事务的service方法中，有数据库操作，有MQ消息的发送，MQ发送后消息消费端需要回查数据库，目前存在的一个情况是在数据库操作事务尚未提交的情况下，消息就发送成功了，此时消息消费者回查数据库，数据依然是未更改的状态，导致消息消费失败。

需求：

我们希望能够控制在事务提交成功之后消息再发送

方案：

采用spring事件监听机制，当监听到事务成功提交后，开始发布消息：

1.事件发布者需要实现ApplicationEventPublisherAware接口：

相关详细解释可以参考以下博客：深入理解spring容器内事件发布监听机制

下面直接上代码如下：

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
@Slf4j
@Component
public class SendMqEventPublisher implements ApplicationEventPublisherAware {

    private static ApplicationEventPublisher eventPublisher;

    @Override
    public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
        if (eventPublisher == null) {
            eventPublisher = applicationEventPublisher;
        }
    }

    /**
     * @description 发布事件
     */
    public void publishEvent(BaseEvent event) {
        log.info("===> 发布sendMq事件:  {}", event);
        eventPublisher.publishEvent(event);
    }

    public void publishMqEvent(String topic, String tag, MessageObject messageObject) {
        SendMqParam sendMqParam = new SendMqParam();
        sendMqParam.setTopic(topic);
        sendMqParam.setTag(tag);
        sendMqParam.setMessageObject(messageObject);
        this.publishEvent(new BaseEvent(sendMqParam));
    }

}

其中要发布的事件体BaseEvent继承自ApplicationEvent：

1
2
3
public  class BaseEvent<T> extends ApplicationEvent {
   //具体消息体自定义
}

2.在事务方法中进行事件发布：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
@Override
   @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW)
   public void processOneAchieve(UpgradeStoreAchievementEntity storeAchievementEntity) {
       	
       	//其他数据库操作相关的业务逻辑
       
       	//事件发布
           UpgradeOperateMqo upgradeOperateMqo = new UpgradeOperateMqo();
           upgradeOperateMqo.setStarstoreId(currentStarstoreId);
           sendMqEventPublisher.publishMqEvent(MessageConstant.STARSTORE_RELATION_TOPIC,
                   MessageConstant.STARSTORE_UPGRADE_OPERATE_TAG, MessageBuilder.overMessage(upgradeOperateMqo));
       } finally {
           distLockSservice.unlock(lockResult);
       }

   }

3.事件监听者通过注解监听发布的事件

通过在方法上增加注解@TransactionalEventListener进行事件监听消费（该注解只监听事务相关的事件），其中参数phase参数可以控制监听到事件后处理事件与提交事务的前后关系：

官方文档也有说明,引用如下：

1.7. Transaction-bound Events

As of Spring 4.2, the listener of an event can be bound to a phase of the transaction. The typical example is to handle the event when the transaction has completed successfully. Doing so lets events be used with more flexibility when the outcome of the current transaction actually matters to the listener.

You can register a regular event listener by using the @EventListener annotation. If you need to bind it to the transaction, use @TransactionalEventListener. When you do so, the listener is bound to the commit phase of the transaction by default.

The next example shows this concept. Assume that a component publishes an order-created event and that we want to define a listener that should only handle that event once the transaction in which it has been published has committed successfully. The following example sets up such an event listener:

Java

1
2
3
4
5
6
7
8
@Component
public class MyComponent {

    @TransactionalEventListener
    public void handleOrderCreatedEvent(CreationEvent<Order> creationEvent) {
        // ...
    }
}

The @TransactionalEventListener annotation exposes a phase attribute that lets you customize the phase of the transaction to which the listener should be bound. The valid phases are BEFORE_COMMIT, AFTER_COMMIT (default), AFTER_ROLLBACK, and AFTER_COMPLETION that aggregates the transaction completion (be it a commit or a rollback).

If no transaction is running, the listener is not invoked at all, since we cannot honor the required semantics. You can, however, override that behavior by setting the fallbackExecution attribute of the annotation to true.

参数值有下面四种：
BEFORE_COMMIT,
AFTER_COMMIT,
AFTER_ROLLBACK,
AFTER_COMPLETION;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@Component
@Slf4j
public class SendMqEventListener {


    @Autowired
    private RocketMQTemplate rocketMQTemplate;


    @TransactionalEventListener(fallbackExecution = true, phase = TransactionPhase.AFTER_COMMIT)
    public void onApplicationEvent(BaseEvent<SendMqParam> event) {
        SendMqParam sendMqParam = (SendMqParam) event.getSource();
        String name = Thread.currentThread().getName();
        log.info("===> 收到sendMq事件:  {}，线程名为： {}", sendMqParam, name);
        rocketMQTemplate.sendNormal(sendMqParam.getTopic(), sendMqParam.getTag(), sendMqParam.getMessageObject());
    }
}

注意spring事件发布者和监听者都需要加入spring管理，@Component注解不要忘记