探索Pulsar之使用Go和Java操作Pulsar

1 前言

Pulsar官方支持的客户端库：

• C++
• Python
• WebSocket
• Go client
• Node.js
• C#
• Java

GitHub中三方的客户端库：

• Go
• Haskell
• Scala
• Rust
• .NET
• Node.js

具体可参看：https://pulsar.apache.org/docs/zh-CN/next/client-libraries/

本次仅演示Go和Java的客户端操作。

2 单机模式运行Pulsar

[root@iZ2ze4m2 bin]# pwd
/root/apache-pulsar-2.10.0/bin
[root@iZ2ze4m2 bin]# ./pulsar standalone

PS:针对单机启动报错问题，如下面的：

可以尝试使用该命令进行启动：

./pulsar standalone -nss

3 Go客户端操作Pulsar

（1）添加依赖

 go get -u "github.com/apache/pulsar-client-go/pulsar"

（2）生产者

package main

import (
   "context"
   "fmt"
   "github.com/apache/pulsar-client-go/pulsar"
   "log"
   "time"
)

func main() {
   client, err := pulsar.NewClient(pulsar.ClientOptions{
      URL:               "pulsar://192.168.71.143:6650", //支持："pulsar://localhost:6650,localhost:6651,localhost:6652"
      OperationTimeout:  60 * time.Second,
      ConnectionTimeout: 60 * time.Second,
   })

   defer client.Close()

   if err != nil {
      log.Fatalf("Could not instantiate Pulsar client: %v", err)
   }

   producer, err := client.CreateProducer(pulsar.ProducerOptions{
      Topic: "my-topic",
   })

   if err != nil {
      log.Fatal(err)
   }

   _, err = producer.Send(context.Background(), &pulsar.ProducerMessage{
      Payload: []byte("hello"),
   })

   defer producer.Close()

   if err != nil {
      fmt.Println("Failed to publish message", err)
   }
   fmt.Println("Published message")

}

（3）消费者

package main

import (
   "context"
   "fmt"
   "github.com/apache/pulsar-client-go/pulsar"
   "log"
   "time"
)

func main() {
   client, err := pulsar.NewClient(pulsar.ClientOptions{
      URL:               "pulsar://192.168.71.143:6650", //支持："pulsar://localhost:6650,localhost:6651,localhost:6652"
      OperationTimeout:  60 * time.Second,
      ConnectionTimeout: 60 * time.Second,
   })

   defer client.Close()

   if err != nil {
      log.Fatalf("Could not instantiate Pulsar client: %v", err)
   }

   consumer, err := client.Subscribe(pulsar.ConsumerOptions{
      Topic:            "my-topic",
      SubscriptionName: "my-sub",
      Type:             pulsar.Shared,
   })
   if err != nil {
      log.Fatal(err)
   }
   defer consumer.Close()

   for i := 0; i < 10; i++ {
      msg, err := consumer.Receive(context.Background())
      if err != nil {
         log.Fatal(err)
      }

      fmt.Printf("Received message msgId: %#v -- content: '%s'\n",
         msg.ID(), string(msg.Payload()))

      consumer.Ack(msg)
   }

   if err := consumer.Unsubscribe(); err != nil {
      log.Fatal(err)
   }
}

4 Java&Spring客户端操作Pulsar

4.1 Java客户端

（1）pom依赖

<properties>
    <pulsar.version>2.9.1</pulsar.version>
</properties>
<dependencies>
    <dependency>
        <groupId>org.apache.pulsar</groupId>
        <artifactId>pulsar-client</artifactId>
        <version>${pulsar.version}</version>
    </dependency>
</dependencies>

（2）生产者和消费者

class SbPursarApplicationTests {

    private PulsarClient client;

    private void init() throws PulsarClientException {
        client = PulsarClient.builder()
                .serviceUrl("pulsar://192.168.71.147:6650")
                .build();
    }

    @Test
    void producer() throws Exception {
        init();
        Producer<byte[]> producer = client.newProducer()
                .topic("my-topic")
                .create();
        // 然后你就可以发送消息到指定的broker 和topic上：
        producer.send("My message".getBytes());
        client.close();
    }

    @Test
    void consumer() throws PulsarClientException {
        init();
        Consumer consumer = client.newConsumer()
                .topic("my-topic")
                .subscriptionName("my-subscription")
                .subscribe();
        while (true) {
            Message msg = consumer.receive();
            try {
                System.out.println("Message received: " + new String(msg.getData()));
                //消息确认
                consumer.acknowledge(msg);
            } catch (Exception e) {
                consumer.negativeAcknowledge(msg);
            }
        }
    }

}

4.2 Spring客户端

（1）依赖

<properties>
    <java.version>1.8</java.version>
    <!-- in your <properties> block -->
    <pulsar.version>2.9.1</pulsar.version>
</properties>
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!-- in your <dependencies> block -->
    <dependency>
        <groupId>org.apache.pulsar</groupId>
        <artifactId>pulsar-client</artifactId>
        <version>${pulsar.version}</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

（2）项目结构

（3）配置类

主要用于将自定义Bean放入Spring中

@Configuration
public class PulsarConfig {

    @Bean
    public Producer pulsarProducer() throws PulsarClientException {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://192.168.71.147:6650")
                .build();
        Producer<byte[]> producer = client
                .newProducer()
                .topic("my-topic")
                .create();
        return producer;
    }

    @Bean
    public Consumer pulsarConsumer() throws PulsarClientException {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://192.168.71.147:6650")
                .build();
        Consumer consumer = client.newConsumer()
                .topic("my-topic")
                .subscriptionName("my-subscription")
                .subscribe();
        return consumer;
    }

}

（4）控制器类：生产者

@RestController
public class HelloPulsarController {

    @Autowired
    private Producer pulsarProducer;

    @RequestMapping("/hello/{msg}")
    public String hello(@PathVariable("msg") String msg) {
        try {
            pulsarProducer.send(msg.getBytes());
        } catch (PulsarClientException e) {
            return "发送失败";
        }
        return "发送成功";
    }

}

（3）消费者

直接使用自定义Bean，并在Spring Boot启动后自动调用该方法

@Service
public class PulsarConsumerService implements ApplicationRunner {

    @Autowired
    private Consumer pulsarConsumer;

    public void consumer() throws PulsarClientException {
        while (true) {
            Message msg = pulsarConsumer.receive();
            try {
                System.out.println("Message received: " + new String(msg.getData()));
                pulsarConsumer.acknowledge(msg);
            } catch (Exception e) {
                pulsarConsumer.negativeAcknowledge(msg);
            }
        }
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        consumer();
    }
}

5 通过pulsar-manager搭建可视化管理界面

5.1 下载链接

https://github.com/apache/pulsar-manager#access-pulsar-manager

或

https://pulsar.apache.org/en/download/

5.2 启动并配置

启动

$ wget https://dist.apache.org/repos/dist/release/pulsar/pulsar-manager/pulsar-manager-0.2.0/apache-pulsar-manager-0.2.0-bin.tar.gz
$ tar -zxvf apache-pulsar-manager-0.2.0-bin.tar.gz
$ cd pulsar-manager
$ tar -xvf pulsar-manager.tar
$ cd pulsar-manager
$ cp -r ../dist ui
$ ./bin/pulsar-manager

配置账号密码

$ CSRF_TOKEN=$(curl http://127.0.0.1:7750/pulsar-manager/csrf-token)
$ curl \
    -H "X-XSRF-TOKEN: $CSRF_TOKEN" \
    -H "Cookie: XSRF-TOKEN=$CSRF_TOKEN;" \
    -H 'Content-Type: application/json' \
    -X PUT http://127.0.0.1:7750/pulsar-manager/users/superuser \
    -d '{"name": "admin", "password": "apachepulsar", "description": "test", "email": "username@test.org"}'

5.3 使用

访问http://localhost:9527就可以打开pulsar-manager界面：

参考：

https://pulsar.apache.org/docs/zh-CN/client-libraries-go/

https://pulsar.apache.org/docs/zh-CN/client-libraries-java/

https://blog.csdn.net/ycf921244819/article/details/120907372

https://github.com/apache/pulsar-manager#access-pulsar-manager