SpringBoot怎么整合Pulsar-java教程-PHP中文网

一、添加pom.xml依赖

二、Pulsar 参数类

三、Pulsar 配置类

四、不同消费数据类型的监听器

五、Pulsar的核心服务类

六、Pulsar整合Spring Cloud

首页

Java

java教程

SpringBoot怎么整合Pulsar

WBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWB

May 24, 2023 pm 01:16 PM

springboot pulsar

一、添加pom.xml依赖

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.7.0</version>
</parent>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <dependency>
        <groupId>org.apache.pulsar</groupId>
        <artifactId>pulsar-client</artifactId>
        <version>2.10.0</version>
    </dependency>

    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.24</version>
        <scope>provided</scope>
    </dependency>
</dependencies>

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <configuration>
                <source>8</source>
                <target>8</target>
            </configuration>
        </plugin>
    </plugins>
</build>

登录后复制

二、Pulsar 参数类

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @Author: huangyibo
 * @Date: 2022/5/28 2:32
 * @Description: Pulsar 参数类
 */

@Component
@ConfigurationProperties(prefix = "tdmq.pulsar")
@Data
public class PulsarProperties {

    /**
     * 接入地址
     */
    private String serviceurl;

    /**
     * 命名空间tdc
     */
    private String tdcNamespace;

    /**
     * 角色tdc的token
     */
    private String tdcToken;

    /**
     * 集群name
     */
    private String cluster;

    /**
     * topicMap
     */
    private Map<String, String> topicMap;

    /**
     * 订阅
     */
    private Map<String, String> subMap;

    /**
     * 开关 on:Consumer可用 ||||| off:Consumer断路
     */
    private String onOff;
}

登录后复制

三、Pulsar 配置类

import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Author: huangyibo
 * @Date: 2022/5/28 2:33
 * @Description: Pulsar 配置类
 */

@Configuration
@EnableConfigurationProperties(PulsarProperties.class)
public class PulsarConfig {

    @Autowired
    PulsarProperties pulsarProperties;

    @Bean
    public PulsarClient getPulsarClient() {

        try {
            return PulsarClient.builder()
                    .authentication(AuthenticationFactory.token(pulsarProperties.getTdcToken()))
                    .serviceUrl(pulsarProperties.getServiceurl())
                    .build();
        } catch (PulsarClientException e) {
            System.out.println(e);
            throw new RuntimeException("初始化Pulsar Client失败");
        }
    }

}

登录后复制

四、不同消费数据类型的监听器

import com.yibo.pulsar.pojo.User;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageListener;
import org.springframework.stereotype.Component;

/**
 * @Author: huangyibo
 * @Date: 2022/5/28 2:37
 * @Description:
 */

@Component
public class UserMessageListener implements MessageListener<User> {

    @Override
    public void received(Consumer<User> consumer, Message<User> msg) {
        try {
            User user = msg.getValue();
            System.out.println(user);
            consumer.acknowledge(msg);
        } catch (Exception e) {
            consumer.negativeAcknowledge(msg);
        }
    }
}
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageListener;
import org.springframework.stereotype.Component;

/**
 * @Author: huangyibo
 * @Date: 2022/5/28 2:37
 * @Description:
 */

@Component
public class StringMessageListener implements MessageListener<String> {

    @Override
    public void received(Consumer<String> consumer, Message<String> msg) {
        try {
            System.out.println(msg.getValue());
            consumer.acknowledge(msg);
        } catch (Exception e) {
            consumer.negativeAcknowledge(msg);
        }
    }
}

登录后复制

五、Pulsar的核心服务类

import com.yibo.pulsar.common.listener.StringMessageListener;
import com.yibo.pulsar.common.listener.UserMessageListener;
import com.yibo.pulsar.pojo.User;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.client.impl.schema.AvroSchema;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * @Author: huangyibo
 * @Date: 2022/5/28 2:35
 * @Description: Pulsar的核心服务类
 */

@Component
public class PulsarCommon {

    @Autowired
    private PulsarProperties pulsarProperties;

    @Autowired
    private PulsarClient client;

    @Autowired
    private UserMessageListener userMessageListener;

    @Autowired
    private StringMessageListener stringMessageListener;


    /**
     * 创建一个生产者 
     * @param topic     topic name
     * @param schema    schema方式
     * @param <T>       泛型
     * @return          Producer生产者
     */
    public <T> Producer<T> createProducer(String topic, Schema<T> schema) {

        try {
            return client.newProducer(schema)
                    .topic(pulsarProperties.getCluster() + "/" + pulsarProperties.getTdcNamespace() + "/" + topic)
                    .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
                    .sendTimeout(10, TimeUnit.SECONDS)
                    .blockIfQueueFull(true)
                    .create();
        } catch (PulsarClientException e) {
            throw new RuntimeException("初始化Pulsar Producer失败");
        }
    }


    /**
     * 
     * @param topic             topic name
     * @param subscription      sub name
     * @param messageListener   MessageListener的自定义实现类
     * @param schema            schema消费方式
     * @param <T>               泛型
     * @return                  Consumer消费者
     */
    public <T> Consumer<T> createConsumer(String topic, String subscription,
                                   MessageListener<T> messageListener, Schema<T> schema) {
        try {
            return client.newConsumer(schema)
                    .topic(pulsarProperties.getCluster() + "/" + pulsarProperties.getTdcNamespace() + "/" + topic)
                    .subscriptionName(subscription)
                    .ackTimeout(10, TimeUnit.SECONDS)
                    .subscriptionType(SubscriptionType.Shared)
                    .messageListener(messageListener)
                    .subscribe();
        } catch (PulsarClientException e) {
            throw new RuntimeException("初始化Pulsar Consumer失败");
        }
    }

    
    /**
     * 异步发送一条消息
     * @param message       消息体
     * @param producer      生产者实例
     * @param <T>           消息泛型
     */
    public <T> void sendAsyncMessage(T message, Producer<T> producer) {
        producer.sendAsync(message).thenAccept(msgId -> {
        });
    }
    
    
    /**
     * 同步发送一条消息
     * @param message       消息体
     * @param producer      生产者实例
     * @param <T>           泛型
     * @throws PulsarClientException
     */
    public <T> void sendSyncMessage(T message, Producer<T> producer) throws PulsarClientException {
        MessageId send = producer.send(message);
        System.out.println();
        System.out.println();
        System.out.println();
        System.out.println();
        System.out.println(send);
    }

    
    //-----------consumer-----------
    @Bean(name = "comment-publish-topic-consumer")
    public Consumer<String> getCommentPublishTopicConsumer() {
        return this.createConsumer(pulsarProperties.getTopicMap().get("comment-publish-topic"),
                pulsarProperties.getSubMap().get("comment-publish-topic-test"),
                stringMessageListener, Schema.STRING);
    }


    @Bean(name = "reply-publish-topic-consumer")
    public Consumer<User> getReplyPublishTopicConsumer() {
        return this.createConsumer(pulsarProperties.getTopicMap().get("reply-publish-topic"),
                pulsarProperties.getSubMap().get("reply-publish-topic-test"),
                userMessageListener, AvroSchema.of(User.class));
    }


    //-----------producer-----------
    @Bean(name = "comment-publish-topic-producer")
    public Producer<String> getCommentPublishTopicProducer() {
        return this.createProducer(pulsarProperties.getTopicMap().get("comment-publish-topic"),Schema.STRING);
    }


    @Bean(name = "reply-publish-topic-producer")
    public Producer<User> getReplyPublishTopicProducer() {
        return this.createProducer(pulsarProperties.getTopicMap().get("reply-publish-topic"), AvroSchema.of(User.class));
    }
}

登录后复制

六、Pulsar整合Spring Cloud

后来发现如上代码会导致BUG-> 在更新Nacos配置之后 Consumer会挂掉
经排查发现结果是由于@RefreshScope注解导致，此注解将摧毁Bean，PulsarConsumer和Producer都将被摧毁，只是说Producer将在下⼀次调⽤中完成重启，Consumer则不能重启，因为没有调⽤，那么怎么解决呢?

就是发布系列事件以刷新容器

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;

/**
 * @Author: huangyibo
 * @Date: 2022/5/28 2:34
 * @Description:
 */

@Component
@Slf4j
public class RefreshPulsarListener implements ApplicationListener {

    @Autowired
    ApplicationContext applicationContext;

    @Override
    public void onApplicationEvent(ApplicationEvent event) {

        if (event.getSource().equals("__refreshAll__")) {
            log.info("Nacos配置中心配置修改 重启Pulsar====================================");
            log.info("重启PulsarClient,{}", applicationContext.getBean("getPulsarClient"));
            log.info("重启PulsarConsumer,{}", applicationContext.getBean("comment-publish-topic-consumer"));
            log.info("重启PulsarConsumer,{}", applicationContext.getBean("reply-publish-topic-consumer"));
        }
    }

}

登录后复制

以上是SpringBoot怎么整合Pulsar的详细内容。更多信息请关注PHP中文网其他相关文章！

本站声明

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系admin@php.cn

热AI工具

热工具

热门话题

gmail邮箱登陆入口在哪里

7546

CakePHP 教程

1382

steam的账户名称是什么格式

win11激活密钥永久

NYT连接提示和答案

显示更多

Related knowledge

Springboot怎么集成Jasypt实现配置文件加密 Jun 01, 2023 am 08:55 AM

Jasypt介绍Jasypt是一个java库，它允许开发员以最少的努力为他/她的项目添加基本的加密功能，并且不需要对加密工作原理有深入的了解用于单向和双向加密的高安全性、基于标准的加密技术。加密密码，文本，数字，二进制文件...适合集成到基于Spring的应用程序中，开放API，用于任何JCE提供程序...添加如下依赖：com.github.ulisesbocchiojasypt-spring-boot-starter2.1.1Jasypt好处保护我们的系统安全，即使代码泄露，也可以保证数据源的

SpringBoot怎么集成Redisson实现延迟队列 May 30, 2023 pm 02:40 PM

使用场景1、下单成功，30分钟未支付。支付超时，自动取消订单2、订单签收，签收后7天未进行评价。订单超时未评价，系统默认好评3、下单成功，商家5分钟未接单，订单取消4、配送超时，推送短信提醒……对于延时比较长的场景、实时性不高的场景，我们可以采用任务调度的方式定时轮询处理。如：xxl-job今天我们采

怎么在SpringBoot中使用Redis实现分布式锁 Jun 03, 2023 am 08:16 AM

一、Redis实现分布式锁原理为什么需要分布式锁在聊分布式锁之前，有必要先解释一下，为什么需要分布式锁。与分布式锁相对就的是单机锁，我们在写多线程程序时，避免同时操作一个共享变量产生数据问题，通常会使用一把锁来互斥以保证共享变量的正确性，其使用范围是在同一个进程中。如果换做是多个进程，需要同时操作一个共享资源，如何互斥呢？现在的业务应用通常是微服务架构，这也意味着一个应用会部署多个进程，多个进程如果需要修改MySQL中的同一行记录，为了避免操作乱序导致脏数据，此时就需要引入分布式锁了。想要实现分

springboot读取文件打成jar包后访问不到怎么解决 Jun 03, 2023 pm 04:38 PM

springboot读取文件，打成jar包后访问不到最新开发出现一种情况，springboot打成jar包后读取不到文件，原因是打包之后，文件的虚拟路径是无效的，只能通过流去读取。文件在resources下publicvoidtest(){Listnames=newArrayList();InputStreamReaderread=null;try{ClassPathResourceresource=newClassPathResource("name.txt");Input

Springboot+Mybatis-plus不使用SQL语句进行多表添加怎么实现 Jun 02, 2023 am 11:07 AM

在Springboot+Mybatis-plus不使用SQL语句进行多表添加操作我所遇到的问题准备工作在测试环境下模拟思维分解一下:创建出一个带有参数的BrandDTO对象模拟对后台传递参数我所遇到的问题我们都知道,在我们使用Mybatis-plus中进行多表操作是极其困难的,如果你不使用Mybatis-plus-join这一类的工具,你只能去配置对应的Mapper.xml文件,配置又臭又长的ResultMap,然后再去写对应的sql语句,这种方法虽然看上去很麻烦,但具有很高的灵活性,可以让我们

SpringBoot与SpringMVC的比较及差别分析 Dec 29, 2023 am 11:02 AM

SpringBoot和SpringMVC都是Java开发中常用的框架，但它们之间有一些明显的差异。本文将探究这两个框架的特点和用途，并对它们的差异进行比较。首先，我们来了解一下SpringBoot。SpringBoot是由Pivotal团队开发的，它旨在简化基于Spring框架的应用程序的创建和部署。它提供了一种快速、轻量级的方式来构建独立的、可执行

SpringBoot怎么自定义Redis实现缓存序列化 Jun 03, 2023 am 11:32 AM

1、自定义RedisTemplate1.1、RedisAPI默认序列化机制基于API的Redis缓存实现是使用RedisTemplate模板进行数据缓存操作的，这里打开RedisTemplate类，查看该类的源码信息publicclassRedisTemplateextendsRedisAccessorimplementsRedisOperations,BeanClassLoaderAware{//声明了key、value的各种序列化方式，初始值为空@NullableprivateRedisSe

springboot怎么获取application.yml里值 Jun 03, 2023 pm 06:43 PM

在项目中，很多时候需要用到一些配置信息，这些信息在测试环境和生产环境下可能会有不同的配置，后面根据实际业务情况有可能还需要再做修改。我们不能将这些配置在代码中写死，最好是写到配置文件中，比如可以把这些信息写到application.yml文件中。那么，怎么在代码里获取或者使用这个地址呢？有2个方法。方法一：我们可以通过@Value注解的${key}即可获取配置文件（application.yml）中和key对应的value值，这个方法适用于微服务比较少的情形方法二：在实际项目中，遇到业务繁琐，逻

See all articles

SpringBoot怎么整合Pulsar

一、添加pom.xml依赖

二、Pulsar 参数类

三、Pulsar 配置类

四、不同消费数据类型的监听器

五、Pulsar的核心服务类

六、Pulsar整合Spring Cloud

热AI工具

Undresser.AI Undress

AI Clothes Remover

Undress AI Tool

Clothoff.io

AI Hentai Generator

热门文章

热工具

记事本++7.3.1

SublimeText3汉化版

禅工作室 13.0.1

Dreamweaver CS6

SublimeText3 Mac版

热门话题