SpringBoot怎么整合Pulsar
一、添加pom.xml依赖org.springframework.bootspring-boot-starter-parent2.7.0org.springframework.bootspring-boot-starter-weborg.apache.pulsarpulsar-client2.10.0org.projectlomboklombok1.18.24providedorg.apache.maven.pluginsmaven-compiler-plugin88二、Pulsar参数类importlo
一、添加pom.xml依赖
org.springframework.boot spring-boot-starter-parent 2.7.0 org.springframework.boot spring-boot-starter-web org.apache.pulsar pulsar-client 2.10.0 org.projectlombok lombok 1.18.24 provided org.apache.maven.plugins maven-compiler-plugin 8 8
二、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 topicMap;
/**
* 订阅
*/
private Map 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{ @Override public void received(Consumer consumer, Message 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 { @Override public void received(Consumer consumer, Message 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 泛型
* @return Producer生产者
*/
public Producer createProducer(String topic, Schema 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 泛型
* @return Consumer消费者
*/
public Consumer createConsumer(String topic, String subscription,
MessageListener messageListener, Schema 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 消息泛型
*/
public void sendAsyncMessage(T message, Producer producer) {
producer.sendAsync(message).thenAccept(msgId -> {
});
}
/**
* 同步发送一条消息
* @param message 消息体
* @param producer 生产者实例
* @param 泛型
* @throws PulsarClientException
*/
public void sendSyncMessage(T message, Producer 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 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 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 getCommentPublishTopicProducer() {
return this.createProducer(pulsarProperties.getTopicMap().get("comment-publish-topic"),Schema.STRING);
}
@Bean(name = "reply-publish-topic-producer")
public Producer 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"));
}
}
} 
韩国外设品牌Pulsar与Noctua联名鼠标FeinmannF01NoctuaEdition于7月21日发售。内置4010规格风扇，支持PWM智能调速；镂空碳纤维外壳，仅重73克。搭载Nordic主控及XS-2传感器，支持42000DPI和8kHz轮询率，性能强劲。联名版采用独特配色与Noctua标志性设计，带来极致操控体验。
Pulsar X3 CrazyLight Medium 鼠标（右手）国行上市：43g，999 元 4月底，外设圈传来新消息。知名品牌Pulsar（派世）旗下的轻量化鼠标X3 CrazyLight，其中号右手版已于本月24日正式登陆国行市场。官方定价999元，并提供了深邃黑（Jet Black）与乌尤
Java操作Pulsar的函数计算是通过编写Java函数在Pulsar集群中处理数据流，以结合Java生态优势和Pulsar的高吞吐、低延迟特性。1.首先搭建Pulsar集群和Java开发环境；2.在Maven项目中引入PulsarFunctionsSDK依赖；3.编写实现Function接口的Java类并实现process方法；4.使用Maven编译打包生成JAR文件；5.通过PulsarCLI部署函数；6.向输入Topic发送消息进行测试。PulsarFunctions还支持Python和Go，监控可
springBoot项目常用目录springBoot项目的目录结构及名规范介绍基于SpringBoot开发时的目录结构及命名规范，通过介绍能够为你解决，在实际项目中如何规划目录结构？如何命名目录更规范？各个目录分别代表什么含义？等三个问题。目录说明servicex//项目名|-admin-ui//管理服务前端代码(一般将UI和SERVICE放到一个工程中，便于管理)|-servicex-auth//模块1|-servicex-common//模块2|-servicex-gateway//模块3|-serv
NVIDIA和MediaTek正携手合作，以便全球游戏玩家更容易地使用业界最强大的游戏显示技术。两家公司今天在Gamescom科隆游戏展上宣布合作，将整套NVIDIAG-SYNC技术集成到全球最热门的scaler组件当中。游戏玩家的显示器：卓越的视觉体验游戏玩家的显示器将提供出色的图像质量、卓越的运动清晰度、超低延迟、高精度的色彩以及更多前沿优势。G-SYNCPulsar：领先的显示技术此次合作的主要亮点是引入G-SYNCPulsar技术，它可提供4倍的有效运动清晰度以及流畅且无撕裂的可变刷新率(VRR)
解析SpringBoot和SpringMVC之间的异同SpringBoot和SpringMVC是Java领域中非常重要的开发框架。虽然它们都属于Spring框架的一部分，但是在使用和功能上有一些明显的区别。本文将对SpringBoot和SpringMVC进行比较，解析它们之间的异同。首先，让我们来了解一下SpringBoot。SpringBo
近年来，随着云计算和分布式架构的兴起，微服务架构的应用越来越广泛。而SpringCloud和SpringBoot作为Java开发中的两个重要框架，对于微服务的实现起到了重要的作用。然而，很多人对于它们在微服务领域的不同应用方式还存在一定的疑惑。本文将从不同的角度探索SpringCloud和SpringBoot在微服务中的应用方式。首先，我们来了解一下Spri
SpringBoot与SpringMVC的不同之处在哪里？SpringBoot和SpringMVC是两个非常流行的Java开发框架，用于构建Web应用程序。尽管它们经常分别被使用，但它们之间的不同之处也是很明显的。首先，SpringBoot可以被看作是一个Spring框架的扩展或者增强版。它旨在简化Spring应用程序的初始化和配置过程，以帮助开发人
SpringBoot和SpringMVC都是Java开发中常用的框架，但它们之间有一些明显的差异。本文将探究这两个框架的特点和用途，并对它们的差异进行比较。首先，我们来了解一下SpringBoot。SpringBoot是由Pivotal团队开发的，它旨在简化基于Spring框架的应用程序的创建和部署。它提供了一种快速、轻量级的方式来构建独立的、可执行
SpringBoot和SpringMVC是Java开发中常用的两个框架，它们都是由Spring框架所提供的，但在功能和使用方式上有着一些区别。本文将分别介绍SpringBoot和SpringMVC的特点和区别。一、SpringBoot的特点：简化配置：SpringBoot通过约定优于配置的原则，大大简化了项目的配置过程。它可以自动配置项目所需要的参数，开发人
火山引擎是字节跳动旗下企业级云与AI服务平台。聚焦大模型核心能力，依托字节自研豆包多模态大模型，凭借火山方舟一站式平台，提供模型精调、推理部署、智能应用搭建等全链路服务，结合高性能AI算力，助力各行业快速落地大模型数字化与智能化应用。
