SpringBoot怎麼整合Pulsar-java教程-PHP中文網

首頁

Java

java教程

SpringBoot怎麼整合Pulsar

WBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWB

May 24, 2023 pm 01:16 PM

springbootpulsar

一、新增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刪除

如何將Maven或Gradle用於高級Java項目管理，構建自動化和依賴性解決方案？Mar 17, 2025 pm 05:46 PM

本文討論了使用Maven和Gradle進行Java項目管理，構建自動化和依賴性解決方案，以比較其方法和優化策略。

如何使用適當的版本控制和依賴項管理創建和使用自定義Java庫（JAR文件）？Mar 17, 2025 pm 05:45 PM

本文使用Maven和Gradle之類的工具討論了具有適當的版本控制和依賴關係管理的自定義Java庫（JAR文件）的創建和使用。

如何使用咖啡因或Guava Cache等庫在Java應用程序中實現多層緩存？Mar 17, 2025 pm 05:44 PM

本文討論了使用咖啡因和Guava緩存在Java中實施多層緩存以提高應用程序性能。它涵蓋設置，集成和績效優勢，以及配置和驅逐政策管理最佳PRA

如何將JPA（Java持久性API）用於具有高級功能（例如緩存和懶惰加載）的對象相關映射？Mar 17, 2025 pm 05:43 PM

本文討論了使用JPA進行對象相關映射，並具有高級功能，例如緩存和懶惰加載。它涵蓋了設置，實體映射和優化性能的最佳實踐，同時突出潛在的陷阱。[159個字符]

Java的類負載機制如何起作用，包括不同的類載荷及其委託模型？Mar 17, 2025 pm 05:35 PM

Java的類上載涉及使用帶有引導，擴展程序和應用程序類負載器的分層系統加載，鏈接和初始化類。父代授權模型確保首先加載核心類別，從而影響自定義類LOA

See all articles

熱AI工具

Undresser.AI Undress

人工智慧驅動的應用程序，用於創建逼真的裸體照片

AI Clothes Remover

用於從照片中去除衣服的線上人工智慧工具。

Undress AI Tool

免費脫衣圖片

Clothoff.io

AI脫衣器

AI Hentai Generator

免費產生 AI 無盡。

熱工具

MinGW - Minimalist GNU for Windows

這個專案正在遷移到osdn.net/projects/mingw的過程中，你可以繼續在那裡關注我們。 MinGW：GNU編譯器集合（GCC）的本機Windows移植版本，可自由分發的導入函式庫和用於建置本機Windows應用程式的頭檔；包括對MSVC執行時間的擴展，以支援C99功能。 MinGW的所有軟體都可以在64位元Windows平台上運作。