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");
}
网友评论