springboot集成Pulsar,⽣产者与消费者⽰例代码话不多说，上代码：
1.导⼊pulsar的jar包
<dependency>
<groupId>org.apache.pulsar</groupId>
睡眠空间<artifactId>pulsar-client</artifactId>
<version>2.6.1</version>
</dependency>
2.pulsar配置（⽰例为yml⽂件）
# pulsar配置
pulsar:
# pulsar服务端地址
url: pulsar://192.168.0.1:30000
# 多个topic以逗号分隔
topic: topic1,topic2
# 消费者组第三方力量
小英雄雨来图片
subscription: topicGroup
3.⽣产者代码⽰例
@Component
教学原则有哪些public class TestPulsarProducer {
private static final Logger log = Logger(TestPulsarProducer.class);
@Value("${pulsar.url}")
private String url;
@Value("${pic}")
private String topic;
PulsarClient client = null;
Producer<byte[]> producer = null;
@PostConstruct
public void initPulsar() throws Exception{
//构造Pulsar client
client = PulsarClient.builder()
.rviceUrl(url)
.build();
//创建producer
producer = wProducer()
.topic(topic.split(",")[0])
.enableBatching(true)//是否开启批量处理消息，默认true，需要注意的是enableBatching只在异步发送ndAsync⽣效，同步发送nd失效。因此建议⽣产环 .compressionType(CompressionType.LZ4)//消息压缩（四种压缩⽅式：LZ4，ZLIB，ZSTD，SNAPPY），consumer端不⽤做改动就能消费，开启后⼤约可 .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS) //设置将对发送的消息进⾏批处理的时间段,10ms；可以理解为若该时间段内批处理成功，则⼀个 .ndTimeout(0, TimeUnit.SECONDS)//设置发送超时0s；如果在ndTimeout过期之前服务器没有确认消息，则会发⽣错误。默认30s，设置为0代表⽆限 .batchingMaxMessages(1000)//批处理中允许的最⼤消息数。默认1000
.maxPendingMessages(1000)//设置等待接受来⾃broker确认消息的队列的最⼤⼤⼩，默认1000
.blockIfQueueFull(true)//设置当消息队列中等待的消息已满时，Producer.nd 和 Producer.ndAsync 是否应该block阻塞。默认为fal，达到maxPendin .roundRobinRouterBatchingPartitionSwitchFrequency(10)//向不同partition分发消息的切换频率，默
认10ms，可根据batch情况灵活调整
.batcherBuilder(BatcherBuilder.DEFAULT)//key_Shared模式要⽤KEY_BASED，才能保证同⼀个key的message在⼀个batch⾥
.create();
}
public void ndMsg(String key, String data){
CompletableFuture<MessageId> future = wMessage()
.key(key)
.Bytes()).ndAsync();//异步发送
future.handle((v, ex) -> {
if (ex == null) {
log.info("Message persisted2: {}", data);
} el {
<("发送Pulsar消息失败msg:【{}】 ", data, ex);
}
return null;
});
无名花// future.join();
log.info("Message persisted: {}", data);
}
}
4.消费者代码
@Component
public class AlarmPulsarConsumer {
private static final Logger log = Logger(AlarmPulsarConsumer.class);
@Value("${pulsar.url}")
private String url;
private String url;
@Value("${pic}")
private String topic;
@Value("${pulsar.subscription}")
private String subscription;
private PulsarClient client = null;
private Consumer consumer = null;
/
**
* 使⽤@PostConstruct注解⽤于在依赖关系注⼊完成之后需要执⾏的⽅法上，以执⾏任何初始化
*/
@PostConstruct
public void initPulsar() throws Exception{
try{
//构造Pulsar client
client = PulsarClient.builder()
.rviceUrl(url)
.build();
//创建consumer
consumer = wConsumer()
.topic(topic.split(","))
.subscriptionName(subscription)
.subscriptionType(SubscriptionType.Shared)//指定消费模式，包含：Exclusive，Failover，Shared，Key_Shared。默认Exclusive模式 .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)//指定从哪⾥开始消费还有Latest，valueof可选，默认Latest
.negativeAckRedeliveryDelay(60, TimeUnit.SECONDS)//指定消费失败后延迟多久broker重新发送消息给consumer，默认60s
.subscribe();
// 开始消费
new Thread(()->{
AlarmPulsarConsumer alarmPulsarConsumer = Bean(AlarmPulsarConsumer.class);
try{
alarmPulsarConsumer.start();
}catch(Exception e){
<("消费Pulsar数据异常，停⽌Pulsar连接：", e);
alarmPulsarConsumer.clo();
}
}).start();
}catch(Exception e){
电饼铛煎牛排("Pulsar初始化异常：",e);
throw e;
}
}
private void start() throws Exception{
//消费消息
while (true) {
Message message = ive();
String[] keyArr = Key().split("_");
String jsons = new Data());
if (StringUtils.isNotEmpty(json)) {
try{
wsSend(type, strategyId, camId, camTime, json,depId,alarmType,target);
}catch(Exception e){
<("消费Pulsar数据异常，key【{}】，json【{}】：", Key(), json, e);
}
}
consumer.acknowledge(message);
}
}
/**
/**隋唐英雄主题曲
* 线程池异步处理Pulsar推送的数据
*
* @param camTime
* @param type
* @param camId
* @param json
*/
// @Async("threadPoolTaskExecutor")
public void wsSend(Integer type, Integer strategyId, String camId, Long camTime, String json,Long depId,Integer alarmType,Integer target) { }
public void clo(){
try {
consumer.clo();
} catch (PulsarClientException e) {
镜中佳人<("关闭Pulsar消费者失败：",e);
}
try {
client.clo();
} catch (PulsarClientException e) {
<("关闭Pulsar连接失败：",e);
}
}
}
