需求
最近公司上了pulsar服务，然后我们需要学习pulsar相关的内容。最好的办法就是自己学习pulsar环境的搭建，然后搭建一个pulsar-server.并且自己建立pulsar-client的消费者和生产者，互相调用，测试连通
pulsar-server
使用docker搭建是最方便的。
输入如下命令就可以啦
docker run -it -p 28000:80 -p 28080:8080 -p 26650:6650 apachepulsar/pulsar-standalone
它会去本地建立一个标准的pulsar server，其中各个端口的意义分别是：
80: the port for pulsar dashboard
8080: the http service url for pulsar service
6650: the binary protocol service url for pulsar service
我这边映射到了28000,28080，26650三个端口。
pulsar-client测试之代码结构
如上图所示，有4个文件，
Client是连接的代码
MessageConsumer是单主题订阅（消费者）
MessageConsumerAll是订阅所有主题（消费者）
MessageProducer是发布指定主题（生产者）
pulsar-client测试之Client.java
配置连接信息。0.0.0.0是IP地址，如果你需要使用，请换成你自己的pulsar服务地址
package pulsar.client;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import com.sun.xml.internal.ws.Closeable;
public class Client {
private PulsarClient client;
public Client() throws PulsarClientException {
client = PulsarClient.builder()
.serviceUrl("pulsar://0.0.0.0:26650/")
.build();
}
public void Close() throws PulsarClientException {
client.close();
}
public PulsarClient getPulsarClient(){
return client;
}
}
pulsar-client测试之MessageConsumer.java
单主题订阅，这段代码是演示单主题订阅，打印收到的订阅内容，不关闭连接
package pulsar.client;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionType;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
public class MessageConsumer {
private Client client;
private Consumer consumer;
public MessageConsumer(String topic, String subscription) throws PulsarClientException {
client = new Client();
consumer = createConsumer(topic, subscription);
}
private Consumer createConsumer(String topic, String subscription) throws PulsarClientException {
return client.getPulsarClient().newConsumer().topic(topic).subscriptionName(subscription)
.ackTimeout(10, TimeUnit.SECONDS).subscriptionType(SubscriptionType.Exclusive).subscribe();
}
public void receiveMessage() throws ExecutionException, InterruptedException, PulsarClientException {
/***
* 用来异步获取，保持回话
*/
do {
// Wait for a message
CompletableFuture<Message> msg = consumer.receiveAsync();
System.out.printf("Message received: %s", new String(msg.get().getData()));
// Acknowledge the message so that it can be deleted by the message broker
consumer.acknowledge(msg.get());
} while (true);
}
public String getMessage() throws ExecutionException, InterruptedException, PulsarClientException {
/***
* 获取一次，就关闭会话
*/
// Wait for a message
System.out.printf("Start pulsar");
CompletableFuture<Message> msg = consumer.receiveAsync();
// System.out.printf("Message received: %s", new String(msg.get().getData()));
String result = "topic is: " + msg.get().getTopicName() + ",data is: " + new String(msg.get().getData());
// Acknowledge the message so that it can be deleted by the message broker
consumer.acknowledge(msg.get());
consumer.close();
client.Close();
return result;
}
public static void main(String[] args) throws PulsarClientException, ExecutionException, InterruptedException {
MessageConsumer consumer = new MessageConsumer("topic1", "my-sub");
consumer.receiveMessage();
// String reString = consumer.getMessage();
// System.err.println(reString);
// consumer.client.Close();
}
}
pulsar-client测试之MessageConsumerAll.java
下面这段代码是演示订阅服务器上的所有主题，收到一条消息之后，打印主题和内容，然后关闭连接
package pulsar.client;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionType;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
public class MessageConsumer {
private Client client;
private Consumer consumer;
public MessageConsumer(String topic, String subscription) throws PulsarClientException {
client = new Client();
consumer = createConsumer(topic, subscription);
}
private Consumer createConsumer(String topic, String subscription) throws PulsarClientException {
return client.getPulsarClient().newConsumer().topic(topic).subscriptionName(subscription)
.ackTimeout(10, TimeUnit.SECONDS).subscriptionType(SubscriptionType.Exclusive).subscribe();
}
public void receiveMessage() throws ExecutionException, InterruptedException, PulsarClientException {
/***
* 用来异步获取，保持回话
*/
do {
// Wait for a message
CompletableFuture<Message> msg = consumer.receiveAsync();
System.out.printf("Message received: %s", new String(msg.get().getData()));
// Acknowledge the message so that it can be deleted by the message broker
consumer.acknowledge(msg.get());
} while (true);
}
public String getMessage() throws ExecutionException, InterruptedException, PulsarClientException {
/***
* 获取一次，就关闭会话
*/
// Wait for a message
System.out.printf("Start pulsar");
CompletableFuture<Message> msg = consumer.receiveAsync();
// System.out.printf("Message received: %s", new String(msg.get().getData()));
String result = "topic is: " + msg.get().getTopicName() + ",data is: " + new String(msg.get().getData());
// Acknowledge the message so that it can be deleted by the message broker
consumer.acknowledge(msg.get());
consumer.close();
client.Close();
return result;
}
public static void main(String[] args) throws PulsarClientException, ExecutionException, InterruptedException {
MessageConsumer consumer = new MessageConsumer("topic1", "my-sub");
consumer.receiveMessage();
// String reString = consumer.getMessage();
// System.err.println(reString);
// consumer.client.Close();
}
}
pulsar-client测试之MessageProducer.java
下面这段代码是发布主题和内容到pulsar服务器，发布一次之后，关闭连接
package pulsar.client;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClientException;
import java.util.concurrent.TimeUnit;
public class MessageProducer {
private Client client;
private Producer<byte[]> producer;
public MessageProducer(String topic) throws PulsarClientException {
client = new Client();
producer = createProducer(topic);
}
private Producer<byte[]> createProducer(String topic) throws PulsarClientException {
return client.getPulsarClient().newProducer()
.topic(topic)
.batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
.sendTimeout(10, TimeUnit.SECONDS)
.blockIfQueueFull(true)
.create();
}
public void sendMessage(String message) {
producer.sendAsync(message.getBytes()).thenAccept(msgId -> {
System.out.printf("Message with ID %s successfully sent", msgId);
});
}
public void sendOnce(String message) {
/**
* 发送一次就关闭
*/
try {
producer.send(message.getBytes());
System.out.printf("Message with content %s successfully sent", message);
producer.close();
client.Close();
} catch (PulsarClientException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
// todo add exceptionally().
public void close(Producer<byte[]> producer){
producer.closeAsync()
.thenRun(() -> System.out.println("Producer closed"));
}
public static void main(String[] args) throws PulsarClientException {
MessageProducer producer = new MessageProducer("topic1");
// producer.sendMessage("Hello World ,lalla");
producer.sendOnce("Hello World ,lizhenwei");
}
}
运行效果
生产者console log:
Message with content Hello World ,lizhenwei successfully sent
消费者console log
Start pulsar receive:
topic is: persistent://public/default/topic1,data is: Hello World ,lizhenwei
