草庐IT

TransactionalEventListener使用场景以及实现原理

老鼠AI大米_Java全栈 2023-03-28 原文

主要讲解TransactionalEventListener是怎样工作的?适合在什么场景,能解决哪些问题?以及和EventListener不同之处。

示例

这里举个业务场景,假如我们有个需求,用户创建成功后给用户发送一个邮件。这里有两个事情要做:

创建用户
给用户发送邮件

对于这种需求,我们可能会不假思索的有以下实现。

@Entity
public class User {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String name;
    private String email;

    public User() {}
    ...
    //getters
    //equals and hashcode
}

为User创建个Repository

public interface UserRepository extends JpaRepository<User, Long> {}


@Service
public class EmailService{

  @Transactional
  public void sendEmail(String email) {
    //send email
  }
}


@Service
public class UserService {
  private final EmailService emailService;
  private final UserRepository userRepository;

  public UserService(EmailService emailService, UserRepository userRepository) {
    this.emailService = emailService;
    this.userRepository = userRepository;
  }

  @Transactional
  public User createUser(User user) {
    User newUser = userRepository.save(user);
    emailService.sendEmail(user.getEmail());
    return newUser;
  }
}

对于上面的实现,是最容易实现的,但这种实现是有问题的。我们想一下,这个功能的核心是创建用户,而发送邮件是一个副作用(发送邮件不能影响用户的创建),如果把这两个操作放在一个事务中会有什么问题?其实很明显,如果创建用户时抛出异常,事务回滚,方法提前退出,那么也不会发送邮件,这是正常的。但是下面两个场景是不可接受的:

  • 如果邮件发送失败,事务发生回滚,用户创建失败。
  • 如果邮件发送成功后,事务提交失败,这下就尴尬了,用户收到了邮件,可是用户创建失败。

虽然这些情况出现的概率很小,但作为对自己有要求的程序猿,这是不可容忍的,我们要对自己写的业务负责。
好了,我们对上面的实现做个重构,直接将创建用户和发送邮件的业务代码拆开,使用Spring application event的方式解耦实现。

修改后的Service是这样的

@Service
public class CustomerService {

    private final UserRepository userRepository;
    private final ApplicationEventPublisher applicationEventPublisher;

    public CustomerService(UserRepository userRepository, ApplicationEventPublisher applicationEventPublisher) {
        this.userRepository = userRepository;
        this.applicationEventPublisher = applicationEventPublisher;
    }

    @Transactional
    public Customer createCustomer(User user) {
        User newUser = userRepository.save(user);
        final UserCreatedEvent event = new UserCreatedEvent(newUser);
        applicationEventPublisher.publishEvent(event);
        return newUser;
    }
}

从上面的代码,我们知道UserService依赖两个beans:

UserRepository - 做持久化工作

ApplicationEventPublisher - 发送Spring内部事件

public class UserCreatedEvent {
private final User user;
public UserCreatedEvent(User user) {
this.user = user;
}
public User getUser() {
return user;
}
...
//equals and hashCode
}

注意这个类只是个简单POJO对象,自从Spring 4.2,我们不用继承ApplicationEvent而能发布任何对象,Spring会把它们包装成PayloadApplicationEvent。
我们需要一个Event Listener处理上面的事件。

@Component
public class UserCreatedEventListener {

    private final EmailService emailService;

    public UserCreatedEventListener(EmailService emailService) {
        this.emailService = emailService;
    }

    @EventListener
    public void processUserCreatedEvent(UserCreatedEvent event) {
        emailService.sendEmail(event.getUser().getEmail());
    }
}

通过上面的重构,我们将创建用户和发送邮件的业务代码拆开来了,但是有解决上面提到的问题吗?答案是没有,虽然我们用EventListener的方式解耦了业务代码,可是这在底层两个功能还是在同一个事务中执行(有人可能想问在Listener方法上加@Async让异步执行可以吗?当然不行,邮件必须在用户创建成功后发送,这里有业务依赖),意思就是,上面的两种情况依然会发生。那么问题来了,有没有解决方案呢?
当然有,就是用@TransactionalEventListener替换@EventListener,结果就是在创建用户并提交事务后发送邮件通知。

TransactionalEventListener

TransactionalEventListener是对EventListener的增强,被注解的方法可以在事务的不同阶段去触发执行,如果事件未在激活的事务中发布,除非显式设置了 fallbackExecution() 标志为true,否则该事件将被丢弃;如果事务正在运行,则根据其 TransactionPhase 处理该事件。
Notice:你可以通过注解@Order去排序所有的Listener,确保他们按自己的设定的预期顺序执行。

如果事件的发布不是在事务(@Transactional)范围内,则监听不到该事件,
除非将fallbackExecution标志设置为true(@TransactionalEventListener(fallbackExecution = true));
如果在事务中,可以选择在事务的哪个阶段来监听事件,默认在事务提交后监听。
修改监听事务的范围:@TransactionalEventListener(phase = TransactionPhase.AFTER_COMPLETION)

我们先看看TransactionPhase有哪些:

  • AFTER_COMMIT - 默认设置,在事务提交后执行
  • AFTER_ROLLBACK - 在事务回滚后执行
  • AFTER_COMPLETION - 在事务完成后执行(不管是否成功)
  • BEFORE_COMMIT - 在事务提交前执行

改造后的Listener是这样的

@Component
public class UserCreatedEventListener {

    private final EmailService emailService;

    public UserCreatedEventListener(EmailService emailService) {
        this.emailService = emailService;
    }

    @TransactionalEventListener
    public void processUserCreatedEvent(UserCreatedEvent event) {
        emailService.sendEmail(event.getUser().getEmail());
    }
}

好了,现在我们能保证我们的业务正常的运行,创建用户不会受发送邮件的影响。接下来我们深挖一下,看看TransactionalEventListener是怎么做到的。
原理分析
下面给出Spring的处理源码,大家会一目了然:

@Override
    public void onApplicationEvent(ApplicationEvent event) {
        if (TransactionSynchronizationManager.isSynchronizationActive() &&
                TransactionSynchronizationManager.isActualTransactionActive()) {
            TransactionSynchronization transactionSynchronization = createTransactionSynchronization(event);
            TransactionSynchronizationManager.registerSynchronization(transactionSynchronization);
        }
        else if (this.annotation.fallbackExecution()) {
            if (this.annotation.phase() == TransactionPhase.AFTER_ROLLBACK && logger.isWarnEnabled()) {
                logger.warn("Processing " + event + " as a fallback execution on AFTER_ROLLBACK phase");
            }
            processEvent(event);
        }
        else {
            // No transactional event execution at all
            if (logger.isDebugEnabled()) {
                logger.debug("No transaction is active - skipping " + event);
            }
        }
    }

解释一下上面的代码:

  • 如果当前处于激活的事务当中,那么会创建一个TransactionSynchronization,并把它放到一个集合当中。意思就是先不执行,只是临时存了起来。
  • 如果没有事务,并且明确设置了fallbackExecution为true,那么直接执行,该效果和EventListener一样。
  • 如果没有事务,并且fallbackExecution 为false,那么直接丢弃该Event不做任何处理。

既然将TransactionSynchronization存放了起来,那么什么时机触发执行呢?
这里以AFTER_COMMIT为例(其他阶段实现差不多),看这段代码:
AbstractPlatformTransactionManager类

private void processCommit(DefaultTransactionStatus status) throws TransactionException {
        try {
            boolean beforeCompletionInvoked = false;

            try {
                boolean unexpectedRollback = false;
                prepareForCommit(status);
                triggerBeforeCommit(status);
                triggerBeforeCompletion(status);
                beforeCompletionInvoked = true;

                if (status.hasSavepoint()) {
                    if (status.isDebug()) {
                        logger.debug("Releasing transaction savepoint");
                    }
                    unexpectedRollback = status.isGlobalRollbackOnly();
                    status.releaseHeldSavepoint();
                }
                else if (status.isNewTransaction()) {
                    if (status.isDebug()) {
                        logger.debug("Initiating transaction commit");
                    }
                    unexpectedRollback = status.isGlobalRollbackOnly();
                    doCommit(status);
                }
                else if (isFailEarlyOnGlobalRollbackOnly()) {
                    unexpectedRollback = status.isGlobalRollbackOnly();
                }

                // Throw UnexpectedRollbackException if we have a global rollback-only
                // marker but still didn't get a corresponding exception from commit.
                if (unexpectedRollback) {
                    throw new UnexpectedRollbackException(
                            "Transaction silently rolled back because it has been marked as rollback-only");
                }
            }
            catch (UnexpectedRollbackException ex) {
                // can only be caused by doCommit
                triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
                throw ex;
            }
            catch (TransactionException ex) {
                // can only be caused by doCommit
                if (isRollbackOnCommitFailure()) {
                    doRollbackOnCommitException(status, ex);
                }
                else {
                    triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
                }
                throw ex;
            }
            catch (RuntimeException | Error ex) {
                if (!beforeCompletionInvoked) {
                    triggerBeforeCompletion(status);
                }
                doRollbackOnCommitException(status, ex);
                throw ex;
            }

            // Trigger afterCommit callbacks, with an exception thrown there
            // propagated to callers but the transaction still considered as committed.
            try {
                triggerAfterCommit(status);
            }
            finally {
                triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
            }

        }
        finally {
            cleanupAfterCompletion(status);
        }
    }

接着看triggerAfterCommit的实现

/**
     * Trigger {@code afterCommit} callbacks.
     * @param status object representing the transaction
     */
    private void triggerAfterCommit(DefaultTransactionStatus status) {
        if (status.isNewSynchronization()) {
            if (status.isDebug()) {
                logger.trace("Triggering afterCommit synchronization");
            }
            TransactionSynchronizationUtils.triggerAfterCommit();
        }
    }

这里调用了TransactionSynchronizationUtils的triggerAfterCommit方法,继续往下跟

  public static void triggerAfterCommit() {
        invokeAfterCommit(TransactionSynchronizationManager.getSynchronizations());
    }

  public static void invokeAfterCommit(@Nullable List<TransactionSynchronization> synchronizations) {
        if (synchronizations != null) {
            for (TransactionSynchronization synchronization : synchronizations) {
                synchronization.afterCommit();
            }
        }
    }

看到了吧,先是拿到所有的TransactionSynchronization,然后调用他们的afterCommit方法,就会真正开始处理该Event。

总结

现在我们做一个总结,如果你遇到这样的业务,操作B需要在操作A事务提交后去执行,那么TransactionalEventListener是一个很好地选择。这里需要特别注意的一个点就是:当B操作有数据改动并持久化时,并希望在A操作的AFTER_COMMIT阶段执行,那么你需要将B事务声明为PROPAGATION_REQUIRES_NEW。这是因为A操作的事务提交后,事务资源可能仍然处于激活状态,如果B操作使用默认的PROPAGATION_REQUIRED的话,会直接加入到操作A的事务中,但是这时候事务A是不会再提交,结果就是程序写了修改和保存逻辑,但是数据库数据却没有发生变化,解决方案就是要明确的将操作B的事务设为PROPAGATION_REQUIRES_NEW。

有关TransactionalEventListener使用场景以及实现原理的更多相关文章

  1. ruby - 如何使用 Nokogiri 的 xpath 和 at_xpath 方法 - 2

    我正在学习如何使用Nokogiri,根据这段代码我遇到了一些问题:require'rubygems'require'mechanize'post_agent=WWW::Mechanize.newpost_page=post_agent.get('http://www.vbulletin.org/forum/showthread.php?t=230708')puts"\nabsolutepathwithtbodygivesnil"putspost_page.parser.xpath('/html/body/div/div/div/div/div/table/tbody/tr/td/div

  2. ruby - 使用 RubyZip 生成 ZIP 文件时设置压缩级别 - 2

    我有一个Ruby程序,它使用rubyzip压缩XML文件的目录树。gem。我的问题是文件开始变得很重,我想提高压缩级别,因为压缩时间不是问题。我在rubyzipdocumentation中找不到一种为创建的ZIP文件指定压缩级别的方法。有人知道如何更改此设置吗?是否有另一个允许指定压缩级别的Ruby库? 最佳答案 这是我通过查看ruby​​zip内部创建的代码。level=Zlib::BEST_COMPRESSIONZip::ZipOutputStream.open(zip_file)do|zip|Dir.glob("**/*")d

  3. ruby - 为什么我可以在 Ruby 中使用 Object#send 访问私有(private)/ protected 方法? - 2

    类classAprivatedeffooputs:fooendpublicdefbarputs:barendprivatedefzimputs:zimendprotecteddefdibputs:dibendendA的实例a=A.new测试a.foorescueputs:faila.barrescueputs:faila.zimrescueputs:faila.dibrescueputs:faila.gazrescueputs:fail测试输出failbarfailfailfail.发送测试[:foo,:bar,:zim,:dib,:gaz].each{|m|a.send(m)resc

  4. ruby-on-rails - 使用 Ruby on Rails 进行自动化测试 - 最佳实践 - 2

    很好奇,就使用ruby​​onrails自动化单元测试而言,你们正在做什么?您是否创建了一个脚本来在cron中运行rake作业并将结果邮寄给您?git中的预提交Hook?只是手动调用?我完全理解测试,但想知道在错误发生之前捕获错误的最佳实践是什么。让我们理所当然地认为测试本身是完美无缺的,并且可以正常工作。下一步是什么以确保他们在正确的时间将可能有害的结果传达给您? 最佳答案 不确定您到底想听什么,但是有几个级别的自动代码库控制:在处理某项功能时,您可以使用类似autotest的内容获得关于哪些有效,哪些无效的即时反馈。要确保您的提

  5. ruby - 在 Ruby 中使用匿名模块 - 2

    假设我做了一个模块如下:m=Module.newdoclassCendend三个问题:除了对m的引用之外,还有什么方法可以访问C和m中的其他内容?我可以在创建匿名模块后为其命名吗(就像我输入“module...”一样)?如何在使用完匿名模块后将其删除,使其定义的常量不再存在? 最佳答案 三个答案:是的,使用ObjectSpace.此代码使c引用你的类(class)C不引用m:c=nilObjectSpace.each_object{|obj|c=objif(Class===objandobj.name=~/::C$/)}当然这取决于

  6. ruby - 使用 ruby​​ 和 savon 的 SOAP 服务 - 2

    我正在尝试使用ruby​​和Savon来使用网络服务。测试服务为http://www.webservicex.net/WS/WSDetails.aspx?WSID=9&CATID=2require'rubygems'require'savon'client=Savon::Client.new"http://www.webservicex.net/stockquote.asmx?WSDL"client.get_quotedo|soap|soap.body={:symbol=>"AAPL"}end返回SOAP异常。检查soap信封,在我看来soap请求没有正确的命名空间。任何人都可以建议我

  7. python - 如何使用 Ruby 或 Python 创建一系列高音调和低音调的蜂鸣声? - 2

    关闭。这个问题是opinion-based.它目前不接受答案。想要改进这个问题?更新问题,以便editingthispost可以用事实和引用来回答它.关闭4年前。Improvethisquestion我想在固定时间创建一系列低音和高音调的哔哔声。例如:在150毫秒时发出高音调的蜂鸣声在151毫秒时发出低音调的蜂鸣声200毫秒时发出低音调的蜂鸣声250毫秒的高音调蜂鸣声有没有办法在Ruby或Python中做到这一点?我真的不在乎输出编码是什么(.wav、.mp3、.ogg等等),但我确实想创建一个输出文件。

  8. ruby-on-rails - 'compass watch' 是如何工作的/它是如何与 rails 一起使用的 - 2

    我在我的项目目录中完成了compasscreate.和compassinitrails。几个问题:我已将我的.sass文件放在public/stylesheets中。这是放置它们的正确位置吗?当我运行compasswatch时,它不会自动编译这些.sass文件。我必须手动指定文件:compasswatchpublic/stylesheets/myfile.sass等。如何让它自动运行?文件ie.css、print.css和screen.css已放在stylesheets/compiled。如何在编译后不让它们重新出现的情况下删除它们?我自己编译的.sass文件编译成compiled/t

  9. ruby - 使用 ruby​​ 将 HTML 转换为纯文本并维护结构/格式 - 2

    我想将html转换为纯文本。不过,我不想只删除标签,我想智能地保留尽可能多的格式。为插入换行符标签,检测段落并格式化它们等。输入非常简单,通常是格式良好的html(不是整个文档,只是一堆内容,通常没有anchor或图像)。我可以将几个正则表达式放在一起,让我达到80%,但我认为可能有一些现有的解决方案更智能。 最佳答案 首先,不要尝试为此使用正则表达式。很有可能你会想出一个脆弱/脆弱的解决方案,它会随着HTML的变化而崩溃,或者很难管理和维护。您可以使用Nokogiri快速解析HTML并提取文本:require'nokogiri'h

  10. ruby - 在 64 位 Snow Leopard 上使用 rvm、postgres 9.0、ruby 1.9.2-p136 安装 pg gem 时出现问题 - 2

    我想为Heroku构建一个Rails3应用程序。他们使用Postgres作为他们的数据库,所以我通过MacPorts安装了postgres9.0。现在我需要一个postgresgem并且共识是出于性能原因你想要pggem。但是我对我得到的错误感到非常困惑当我尝试在rvm下通过geminstall安装pg时。我已经非常明确地指定了所有postgres目录的位置可以找到但仍然无法完成安装:$envARCHFLAGS='-archx86_64'geminstallpg--\--with-pg-config=/opt/local/var/db/postgresql90/defaultdb/po

随机推荐