SpringBoot整合Pulsar的实现示例
更新时间：2022年07月01日 16:33:32 作者：小波同学
本文主要介绍了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"));
}
}
}参考：
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内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
GsonFormat主要用于使用Gson库将JSONObject格式的String 解析成实体，本文主要介绍了GsonFormat快速生成JSon实体类的实现，具有一定的参考价值，感兴趣的可以了解一下2023-05-05
springboot ElasticSearch如何配置自定义转换器ElasticsearchCustomConver
这篇文章主要介绍了springboot ElasticSearch如何配置自定义转换器ElasticsearchCustomConversions问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2023-08-08
这篇文章主要介绍了idea中项目前端网页图标不显示的原因及解决方案，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2023-07-07
这篇文章主要给大家介绍了关于如何利用Java简单实现一个代码行数统计器的相关资料，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2020-11-11
这篇文章主要介绍了从JVM角度对MyEclipse进行调试优化，为大家分析调试优化MyEclipse的步骤，感兴趣的小伙伴们可以参考一下2016-05-05
本文主要介绍了springboot处理异常的5种方式，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2022-08-08
spring boot整合mybatis使用c3p0数据源连接mysql
这篇文章主要为大家详细介绍了spring boot整合mybatis使用c3p0数据源连接mysql，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2018-03-03
这篇文章主要为大家详细介绍了java实现上传图片尺寸修改和质量压缩，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2022-04-04
SpringBoot之如何搭建SpringBoot+Maven项目
这篇文章主要介绍了SpringBoot之如何搭建SpringBoot+Maven项目问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2024-07-07
这篇文章主要为大家介绍了elasticsearch分布式及数据功能源码分析，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪2022-04-04
最新评论