codeorg免费编程网站,网站开发学校有哪些,微信云开发平台,做网站 就上微赞网译者信息 王中兴#xff0c;就职于 ebay 消息中间件团队#xff0c;社区昵称 alphawang。 我们很高兴地宣布 Spring for Apache Pulsar[1] 的第一个里程碑版本已发布#xff0c;现在你可以在 Spring 应用程序里直接利用 Apache Pulsar 的强大能力。 我们先来看一下 Apache P… 译者信息 王中兴就职于 ebay 消息中间件团队社区昵称 alphawang。 我们很高兴地宣布 Spring for Apache Pulsar[1] 的第一个里程碑版本已发布现在你可以在 Spring 应用程序里直接利用 Apache Pulsar 的强大能力。 我们先来看一下 Apache Pulsar 与 Spring 集成带来的好处然后通过示例应用程序讲解如何进行集成。 为什么使用 Spring for Apache Pulsar Spring 是当今最流行的 Java 框架它帮助开发人员快速、安全、轻松地创建生产就绪的应用程序。它是一个灵活的框架提供开箱即用的默认设置以提高开发效率还支持针对随时出现的需求进行定制。这使其成为构建云原生应用程序时的理想选择。 Apache Pulsar 是一个云原生流和消息平台使组织能够在弹性云环境中构建可扩展、可靠的应用程序。它结合了传统消息系统以及发布订阅系统的最佳特性。在 Pulsar 的多层架构中每一层都是可扩展、分布式的并且与其他层解耦。计算和存储的分离支持独立地扩展每一层。 Pulsar 和 Spring 结合在一起可以让你轻松快速地构建可扩展、稳健的数据应用程序。将 Pulsar 与 Spring 微服务相整合可以进一步实现与其他语言编写的服务的无缝互操作。Spring for Apache Pulsar 提供了一个与 Pulsar 交互的工具包。从模板到监听器和自动配置所有你喜欢的 Spring 概念现在都可以与 Pulsar 一起使用。如果你正在使用 Spring for Kafka 或 Spring AMQP那么将 Pulsar 融入到你的现有架构中则非常简单。Spring for Pulsar 采用了相同的概念使用起来不会有陌生的感觉。 如何使用 Spring for Apache Pulsar 我们将构建一个示例应用程序通过消费注册数据来提醒客户成功团队有新客户。Spring 将运行我们的应用程序并提供配置而 Pulsar 则作为消息总线来路由我们的数据。 该示例程序的完整源代码参考 GitHub 仓库[2] 。 观看如下演示视频了解 Spring for Apache Pulsar 的实际应用 先决条件 示例应用程序使用 Maven 和 Java 17 运行。要开始使用 Spring for Apache Pulsar首先需要将其作为依赖项添加到我们的 Spring 项目中。 propertiesspring-pulsar.version0.1.0-M1/spring-pulsar.version
/properties
dependenciesdependencygroupIdorg.springframework.pulsar/groupIdartifactIdspring-pulsar-spring-boot-starter/artifactIdversion${spring-pulsar.version}/version/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-configuration-processor/artifactId/dependency
/dependencies
buildpluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/plugins
/build
repositoriesrepositoryidspring-milestones/idnameSpring Milestones/nameurlhttps://repo.spring.io/milestone/urlsnapshotsenabledfalse/enabled/snapshots/repository
/repositories
pluginRepositoriespluginRepositoryidspring-milestones/idnameSpring Milestones/nameurlhttps://repo.spring.io/milestone/urlsnapshotsenabledfalse/enabled/snapshots/pluginRepository
/pluginRepositories 执行 mvn clean package 进行编译并执行 mvn spring-boot:run 运行该示例应用程序。 我们还需要一个 Pulsar 集群来运行该程序。我们可以使用本地的 Standalone Pulsar 集群也可以使用 StreamNative Cloud[3] 提供集群。 连接到 Pulsar 接下来使用 Spring Configuration 配置应用程序连接到 Pulsar。将以下内容添加到 src/main/resources/application.yml。 spring:pulsar:client:service-url: pulsarssl://free.o-j8r1u.snio.cloud:6651auth-plugin-class-name: org.apache.pulsar.client.impl.auth.oauth2.AuthenticationOAuth2authentication:private-key: file:///Users/user/Downloads/o-j8r1u-free.jsonaudience: urn:sn:pulsar:o-j8r1u:freeissuer-url: https://auth.streamnative.cloud/producer:batching-enabled: falsetopic-name: persistent://public/default/signups-topic 该配置使用 OAuth2 身份验证连接到 StreamNative Cloud。请参照 StreamNative Cloud 文档[4]获取身份验证凭据。我们还关闭了 Pulsar 生产者的批处理消息功能并设置了其默认主题名称。 生产数据 现在可以开始向我们的集群发送消息了。在本示例中我们将生成模拟的用户注册数据并将其不断写入 signups-topic 主题。通过 Spring for Apache Pulsar直接将 PulsarTemplate 添加到应用程序代码即可发送消息。 EnableScheduling
SpringBootApplication
public class SignupApplication {private static final Logger log LoggerFactory.getLogger(SignupApplication.class);Autowired private PulsarTemplateSignup signupTemplate;Autowired private PulsarTemplateCustomer customerTemplate;Scheduled(initialDelay 5000, fixedRate 5000)void publishSignupData() throws PulsarClientException {Signup signup signupGenerator.generate();signupTemplate.setSchema(JSONSchema.of(Signup.class));signupTemplate.send(signup);}…
} 上面的代码创建了一个定时任务来生成模拟的用户注册信息并将其作为消息发送到我们在 application.yml 中配置的默认主题。我们可以看到通过配置 PulsarTemplate 来使用 JSON Schema 发送消息非常简单。 消费数据 为了从我们的数据中获取价值现在要对传入的注册信息进行过滤。如果注册的 tier 是 ENTERPRISE我们将创建一个新 Customer 对象并向 customer-success 主题发布一条消息。要想消费 signups-topic 主题我们需要在 SignupApplication 类中添加 PulsarListener 注解。 PulsarListener(subscriptionName signup-consumer,topics signups-topic,schemaType SchemaType.JSON)
void filterSignups(Signup signup) throws PulsarClientException {log.info({} {} ({}) just signed up for {} tier,signup.firstName(),signup.lastName(),signup.companyEmail(),signup.signupTier());if (signup.signupTier() SignupTier.ENTERPRISE) {Customer customer Customer.from(signup);customerTemplate.setSchema(JSONSchema.of(Customer.class));customerTemplate.send(customer-success, customer);}
} PulsarListener 注解配置了一个 Pulsar 消费者来使用给定的 Schema 读取特定主题。在 filterSignups 方法中我们使用了之前添加的第二个 PulsarTemplate。这次我们不想向默认主题发送消息因此我们传入 customer-success 作为要写入的主题名称。 最后我们的客户成功团队现在可以收到所有新企业客户注册的提醒。要做到这一点他们只需要使用 Customer Schema 来消费 customer-success 主题即可。 PulsarListener(subscriptionName customer-consumer,topics customer-success,schemaType SchemaType.JSON)
void alertCustomerSuccess(Customer customer) {log.info(## Start the onboarding for {} - {} {} ({}) - {} ##,customer.companyName(),customer.firstName(),customer.lastName(),customer.phoneNumber(),customer.companyEmail());
} 高级特性 Spring for Apache Pulsar 提供了许多高级特性。例如我们可以使用 ProducerInterceptor 来调试消息的记录。 首先我们在 Spring 配置类中添加一个 ProducerInterceptor bean。 ProducerInterceptor 实现类仅需要在 Broker 确认后记录消息信息。 Configuration(proxyBeanMethods false)
class SignupConfiguration {BeanProducerInterceptor loggingInterceptor() {return new LoggingInterceptor();}static class LoggingInterceptor implements ProducerInterceptor {private static final Logger log LoggerFactory.getLogger(LoggingInterceptor.class);Overridepublic void close() {// no-op}Overridepublic boolean eligible(Message message) {return true;}Overridepublic Message beforeSend(Producer producer, Message message) {return message;}Overridepublic void onSendAcknowledgement(Producer producer, Message message, MessageId msgId, Throwable exception) {log.debug(MessageId: {}, Value: {}, message.getMessageId(), message.getValue());}}
} 得益于 Spring 自动配置的特性我们的应用程序将自动配置 LoggingInterceptor 来拦截所有 Pulsar 生产者发送的消息。 接下来在 application.yml 中将日志级别设置为 debug拦截器即可正常工作。 logging:level:io.streamnative.example: debug 总结 在这篇博客中我们探讨了如何使用 Spring for Apache Pulsar 快速构建一个示例应用程序。Spring for Apache Pulsar 集成还提供更多功能例如订阅类型、批处理和手动确认。对于进一步需要外部系统中数据读写权限的高级应用程序我们可以添加 Pulsar IO 连接器。如果你有兴趣了解更多 Pulsar IO 连接器的信息请访问 StreamNative Hub[5]。 更多资源 • 查看用户注册示例应用程序的完整源代码[6]• 免费试用 StreamNative Cloud[7] 以开始使用 Apache Pulsar• 帮助 Spring for Apache Pulsar[8] 开源项目发展• 查看 Spring for Apache Pulsar[9] 文档 引用链接 [1] Spring for Apache Pulsar: https://docs.spring.io/spring-pulsar/docs/0.1.0-M1/reference/html/[2] GitHub 仓库: https://github.com/streamnative/examples/tree/master/spring-pulsar[3] StreamNative Cloud: https://streamnative.io/streamnativecloud/[4] StreamNative Cloud 文档: https://docs.streamnative.io/cloud/stable/connect/overview[5] StreamNative Hub: https://hub.streamnative.io/[6] 完整源代码: https://github.com/streamnative/examples/tree/master/spring-pulsar[7] StreamNative Cloud: https://streamnative.io/streamnativecloud/[8] Spring for Apache Pulsar: https://github.com/spring-projects-experimental/spring-pulsar[9] Spring for Apache Pulsar: https://docs.spring.io/spring-pulsar/docs/0.1.0-M1/reference/html/
