世界已经进入“移动”时代，现在应用程序必须能够实时提供数据，这不仅包括存储在数据库表中的重要最终结果，还包括用户在使用应用程序期间执行的所有操作。任何可用信息，例如用户点击、日志数据或传感器数据，都可用于改善用户体验、生成报告、向机器学习系统提供数据等。今天，开发人员必须专注于基于实时事件流的系统。下图显示了基于事件流处理的体系结构的示例。ApacheKafka已成为构建高度可扩展的基于事件的系统的首选解决方案。Kafka为事件流平台提供快速升级能力，供开发人员在现代业务解决方案中使用。但是，开发人员通常需要将现有的JavaEE业务解决方案（基于IBMMQ或IBMWebSphereApplicationPlatform等技术构建）集成到这些新的事件流架构中。考虑以下示例：在线商店有一个移动应用程序，该应用程序使用Kafka将支付请求数据发送到用EnterpriseJava实现的分布式支付系统。解决方案必须绝对保证付款请求被处理一次（避免向买家多次收费）。然而，在分布式系统中故障是不可避免的，因此解决方案需要以优雅的方式处理故障。使用ApacheKafka实现消息传递ApacheKafka是一种用于事件流处理的分布式系统，广泛用于微服务架构和基于云的环境中。它在同一平台内提供事件的消息传递、存储和处理。下图显示了ApacheKafka组件的基本拓扑结构，包括生产者和消费者通过Kafka集群基础设施交换消息。尽管Kafka有很多优点，但Kafka仍然面临以下问题：消息处理失败时需要手动实现补偿逻辑，可能导致消息无法处理不支持XA事务需要开发和可维护性工作将其集成到一个企业解决方案要解决Kafka集成问题，您可以应用传统的消息拓扑概念，例如事务日志、恢复日志和XA事务。您可以基于JavaEE连接器体系结构(JCA)实现资源适配器。使用这个JCA资源适配器，您可以为应用程序服务器提供用于Kafka消息处理的ACID功能。然后，此JCA资源适配器提供与企业Java应用程序的无缝Kafka集成。实施JCA资源适配器JavaEE连接器体系结构定义了一组可扩展且高度安全的事务处理机制。您可以将JCA资源适配器安装到任何符合JavaEE的应用程序服务器中，例如IBMWebsphereApplicationServer、IBMBusinessProcessManager、JBoss、WebSphereLiberty、Glassfish或Weblogic。JavaEECon??nectorArchitecture规范还提供了一组标准的契约，用于支持企业应用程序和企业信息系统（例如Kafka）之间的通信。JCA资源适配器可以插入应用程序服务器并通过处理所有系统级机制（事务、连接管理、崩溃恢复、错误跟踪和日志记录）来支持Kafka集成。JCA资源适配器将从需要与其集成的企业应用程序中隐藏所有Kafka通信逻辑。通过实施JCA资源适配器，企业应用程序提供商可以专注于实施业务和表示逻辑，而不必担心与Kafka集成相关的低级逻辑。因此，JCA资源适配器只需开发一次，即可被各种应用程序重复使用。让我们将此与在线商店支付场景相关联，并观察下图，其中显示了指定的解决方案系统上下文。移动应用向Kafka发送支付请求数据，Kafka通过资源适配器与企业支付应用集成。此外，此适配器可用于将支付通知推送到Kafka。此适配器启动传递给企业支付应用程序和通知系统的XA事务。因此，与支付请求处理相关的所有任务将在同一个全局事务中运行，并且同时完成或失败。除了读取或写入数据的主题外，设计还设置了重试、死信和事务日志主题。现在，让我们更详细地探讨与移动应用程序之间的消息处理。传入流在我们的支付场景中，传入流表示由网络商店移动应用程序发起的通信，该应用程序将支付请求数据发送到Kafka。资源适配器提供Kafka连接并将消息异步传递到应用程序服务器上存在的消息端点。这是使用JCA规范定义的消息流入协定实现的。KafkaJCA资源适配器实现了一个激活规范JavaBean，它包含一组用于端点激活配置的配置属性。这些配置细节被定义为应用程序服务器配置的一部分。资源适配器定期从传入的Kafka主题中轮询一批支付请求。成功完成数据轮询后，它会遍历成批数据并将消息异步传递到端点实例。每个消息端点可能存在多个端点实例，从而实现消息的并行消费并提供高吞吐量。一旦消息被安排交付，Kafka消费者偏移量就会被提交，从而避免了批处理阻塞问题。这种设计之所以可行，是因为资源适配器实现了具有重试、死信和需要在Kafka上设置的事务日志主题的故障转移过程。在我们的例子中，端点需要支持XA事务，并且在将数据发送到端点之前需要创建事务上下文，从而提供原子消息消费。如果应用服务器中止事务，则应回滚端点实例执行的所有工作，并将消息转发到Kafka重试主题。适配器使用来自Kafka重试主题的消息并重新处理它们。在超过配置的消息处理重试次数后，适配器将消息传递到Kafka死信主题。发送到死信主题的消息包含有价值的业务数据，因此监控该主题至关重要。传出传出流表示由企业应用程序发起的Kafka通信。在我们的例子中，这是用于向移动应用程序发送支付确认的通知系统。JCA规范定义了一个连接管理合同，使应用程序服务器能够整合Kafka连接，提供支持大量客户端的可扩展环境。Kafka传出连接配置细节是使用托管连接工厂JavaBean定义的。使用这些配置细节，管理员和开发人员使用适配器来配置Kafka生产者并决定所需的功能，例如可靠性、可用性、吞吐量、延迟和事务支持。这些配置细节被定义为应用程序服务器配置的一部分。KafkaJCA资源适配器公开了实现通用客户端接口(CCI)和Java消息服务(JMS)接口的Kafka连接工厂和Kafka连接。应用程序组件使用Java命名和目录接口(JNDI)名称查找连接工厂。成功获取连接工厂后，应用程序使用它获取与Kafka的连接。这允许您为当前将数据发送到JMS消息传递提供程序（例如IBMMQ或ActiveMQ）的通知系统应用程序无缝添加Kafka集成。资源适配器传出流封装了低级Kafka通信逻辑并提供：连接合并使用Kafka事务机制来保证只传递一次采用安全的方式来识别、记录和处理Kafka故障通过Kafka提供可靠的消息处理来管理事务在传出流中，Kafka资源适配器可以使用JCA规范定义的事务管理契约。在我们的例子中，需要设置连接工厂以支持XA事务，并且适配器需要在客户端获取连接时启动Kafka事务。每当应用服务器回滚Kafka事务时，事务异常终止。如果发生XA事务提交，事务管理器将对正在运行的事务使用的所有资源执行两阶段提交协议。这保证了对托管资源的所有读写访问都完全提交或完全回滚。最后，资源适配器通过将事务数据写入Kafka事务日志主题来跟踪正在运行的事务。写入事务日志主题的数据将用于崩溃恢复处理，以在分布式系统中提供可靠的消息处理。结论KafkaJCA适配器的设计方法为标准企业Java解决方案提供了JMS与Kafka事件处理平台的“即插即用”集成。这种设计使Kafka与现有企业应用程序无缝集成，而无需实施补偿逻辑。通过这个适配器，应用服务器也提供了企业应用建立Kafka连接和进行事务管理所依赖的基础设施和运行环境。
