spring boot pulsar,

一、添加pom.xml依赖

父groupIdorg.springframework.boot/groupId artifact id spring-boot-starter-parent/artifact id版本2。7 .0/版本/父依赖项依赖项groupIdorg.springframework.boot/groupId artifact id spring-boot-starter-web/artifact id/依赖项groupIdorg.apache.pulsar/groupId artifact id pulsar-client/artifact id版本2。10 .0/版本/依赖项依赖项groupIdorg.projectlombok/groupId artifact id lombok/artifact id版本1。18 .24/提供的版本范围/范围/依赖项构建插件groupIdorg.apache.maven.plugins/groupId人工化的maven-编译器-插件

二、Pulsar 参数类

导入龙目岛。数据；导入org。spring框架。靴子。语境。属性。配置属性；导入org。spring框架。刻板印象。组件；导入Java。util。地图；/* * * * @作者：黄艺博* @ date : 2022/5/28 2:32 * @ description :脉冲星参数类*/@ Component @配置属性(前缀= tdmq。pulsar )@ data公共类pulsar属性{/* * *接入地址*/私有字符串服务url/** *命名空间贸发会议*/私有字符串tdcNamespace/** *角色上止点(上止点的缩写)的令牌*/私有字符串tdcToken/** *集群姓名*/私人字符串簇；/** * topicMap */private MapString，String topicMap/** *订阅*/private MapString，String subMap/** *开关：消费者可用

　　 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: */@Componentpublic 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: */@Componentpublic 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的核心服务类 */@Componentpublic 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@Slf4jpublic 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"));        }    }}

参考：

　　https://wenku.baidu.com/view/4d3337ab6b0203d8ce2f0066f5335a8102d266a7.html

　　https://gitee.com/zhaoyuxuan66/pulsar-springcloud_boot-demo/tree/master/

　　https://blog.csdn.net/weixin_56227932/article/details/122897075

　　http://www.zzvips.com/article/219361.html

　　https://mp.weixin.qq.com/s/4w0eucDNcrYrsiDXHzLwuQ

　　到此这篇关于SpringBoot整合Pulsar的实现示例的文章就介绍到这了,更多相关SpringBoot整合Pulsar内容请搜索盛行IT以前的文章或继续浏览下面的相关文章希望大家以后多多支持盛行IT！

郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。