1. 官网?https://pulsar.apache.org/docs/3.0.x/getting-started-docker/
安装命令：?docker run -it -p 6650:6650 -p 8080:8080 --mount source=pulsardata,target=/pulsar/data --mount source=pulsarconf,target=/pulsar/conf apachepulsar/pulsar:3.0.0 bin/pulsar standalone
2. 安装? pip install pulsar-client? 测试pulsar 安装是否成功
consumer.py? 文件内容
import pulsar
client = pulsar.Client('pulsar://localhost:6650')
consumer = client.subscribe('my-topic', subscription_name='my-sub')
while True:
? ? msg = consumer.receive()
? ? print("Received message: '%s'" % msg.data())
? ? consumer.acknowledge(msg)
client.close()
------------------------------------------------------------------------
product.py 文件内容
import pulsar
client = pulsar.Client('pulsar://localhost:6650')
producer = client.create_producer('my-topic')
for i in range(10):
? ? producer.send(('hello-pulsar-%d' % i).encode('utf-8'))
client.close()
-----------------------------------------------------
查看 topic 信息：curl http://localhost:8080/admin/v2/persistent/public/default/my-topic/stats | python -m json.tool
3. java client 访问 pulsar，?pulsar-client.version = 3.0.0
<dependency>
? ? <groupId>org.apache.pulsar
? ? <artifactId>pulsar-client
? ? <version>${pulsar-client.version}
</dependency>
使用 api 发送
package org.example.liteflow.pulsar;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@Component
@Slf4j
public class PulsarProducer {
@Value("${pulsar.url}")
private Stringurl;
? ? @Value("${pulsar.topic}")
private Stringtopic;
? ? PulsarClientclient =null;
? ? Producerproducer =null;
? ? @PostConstruct
? ? public void initPulsar()throws Exception{
//构造Pulsar client
? ? ? ? client = PulsarClient.builder()
.serviceUrl(url)
.build();
? ? ? ? //创建producer
? ? ? ? producer =client.newProducer()
.topic(topic.split(",")[0])
.enableBatching(true)//是否开启批量处理消息，默认true,需要注意的是enableBatching只在异步发送sendAsync生效，同步发送send失效。因此建议生产环境若想使用批处理，则需使用异步发送，或者多线程同步发送
? ? ? ? ? ? ? ? .compressionType(CompressionType.LZ4)//消息压缩（四种压缩方式：LZ4，ZLIB，ZSTD，SNAPPY），consumer端不用做改动就能消费，开启后大约可以降低3/4带宽消耗和存储（官方测试）
? ? ? ? ? ? ? ? .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)//设置将对发送的消息进行批处理的时间段,10ms；可以理解为若该时间段内批处理成功，则一个batch中的消息数量不会被该参数所影响。
? ? ? ? ? ? ? ? .sendTimeout(0, TimeUnit.SECONDS)//设置发送超时0s；如果在sendTimeout过期之前服务器没有确认消息，则会发生错误。默认30s，设置为0代表无限制，建议配置为0
? ? ? ? ? ? ? ? .batchingMaxMessages(1000)//批处理中允许的最大消息数。默认1000
? ? ? ? ? ? ? ? .maxPendingMessages(1000)//设置等待接受来自broker确认消息的队列的最大大小，默认1000
? ? ? ? ? ? ? ? .blockIfQueueFull(true)//设置当消息队列中等待的消息已满时，Producer.send 和 Producer.sendAsync 是否应该block阻塞。默认为false，达到maxPendingMessages后send操作会报错，设置为true后，send操作阻塞但是不报错。建议设置为true
? ? ? ? ? ? ? ? .roundRobinRouterBatchingPartitionSwitchFrequency(10)//向不同partition分发消息的切换频率，默认10ms，可根据batch情况灵活调整
? ? ? ? ? ? ? ? .batcherBuilder(BatcherBuilder.DEFAULT)//key_Shared模式要用KEY_BASED,才能保证同一个key的message在一个batch里
? ? ? ? ? ? ? ? .create();
? ? }
public void sendMsg(String key, String data) {
CompletableFuture future =producer.newMessage()
.key(key)
.value(data.getBytes()).sendAsync();//异步发送
? ? ? ? future.handle((v, ex) -> {
if (ex ==null) {
log.info("Message persisted2: {}", data);
? ? ? ? ? ? }else {
log.error("发送Pulsar消息失败msg:【{}】 ", data, ex);
? ? ? ? ? ? }
return null;
? ? ? ? });
? ? ? ? // future.join();
? ? ? ? log.info("Message persisted: {}", data);
? ? }
}
消费:
package org.example.liteflow.pulsar;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import java.util.concurrent.TimeUnit;
@Component
@Slf4j
public class PulsarConsumer {
@Value("${pulsar.url}")
private Stringurl;
? ? @Value("${pulsar.topic}")
private Stringtopic;
? ? @Value("${pulsar.subscription}")
private Stringsubscription;
? ? @Resource
? ? private ApplicationContextcontext;
? ? private PulsarClientclient =null;
? ? private Consumerconsumer =null;
? ? /**
? ? * 使用@PostConstruct注解用于在依赖关系注入完成之后需要执行的方法上，以执行任何初始化
? ? */
? ? @PostConstruct
? ? public void initPulsar()throws Exception{
try{
//构造Pulsar client
? ? ? ? ? ? client = PulsarClient.builder()
.serviceUrl(url)
.build();
? ? ? ? ? ? //创建consumer
? ? ? ? ? ? consumer =client.newConsumer()
.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(()->{
PulsarConsumer alarmPulsarConsumer =context.getBean(PulsarConsumer.class);
? ? ? ? ? ? ? ? try{
alarmPulsarConsumer.start();
? ? ? ? ? ? ? ? }catch(Exception e){
log.error("消费Pulsar数据异常，停止Pulsar连接：", e);
? ? ? ? ? ? ? ? ? ? alarmPulsarConsumer.close();
? ? ? ? ? ? ? ? }
}).start();
? ? ? ? }catch(Exception e){
log.error("Pulsar初始化异常：",e);
? ? ? ? ? ? throw e;
? ? ? ? }
}
private void start()throws Exception{
//消费消息
? ? ? ? while (true) {
Message message =consumer.receive();
? ? ? ? ? ? //String[] keyArr = message.getKey().split("_");
? ? ? ? ? ? String json =new String(message.getData());
? ? ? ? ? ? if (StringUtils.hasText(json)) {
try{
log.info("消费Pulsar数据，key【{}】，json【{}】：", message.getKey(), json);
? ? ? ? ? ? ? ? ? ? //wsSend(type, strategyId, camId, camTime, json,depId,alarmType,target);
? ? ? ? ? ? ? ? }catch(Exception e){
log.error("消费Pulsar数据异常，key【{}】，json【{}】：", message.getKey(), 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 close() {
try {
consumer.close();
? ? ? ? }catch (PulsarClientException e) {
log.error("关闭Pulsar消费者失败：", e);
? ? ? ? }
try {
client.close();
? ? ? ? }catch (PulsarClientException e) {
log.error("关闭Pulsar连接失败：", e);
? ? ? ? }
}
}
4. 写一个 controller 的method 发送
@GetMapping("/send/{key}/{data}")
public ResponseEntitysend(@PathVariable("key") String key,@PathVariable("data") String data){
pulsarProducer.sendMsg(key, data);
? ? return ResponseEntity.ok("send successful");
}