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

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的客户端操作。

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

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

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

./pulsar standalone -nss

（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)
   }
}

（1）pom依赖

    2.9.1


    
        org.apache.pulsar
        pulsar-client
        ${pulsar.version}
    

（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 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);
            }
        }
    }

}

（1）依赖

    1.8
    
    2.9.1


    
        org.springframework.boot
        spring-boot-starter
    
    
        org.springframework.boot
        spring-boot-starter-web
    
    
    
        org.apache.pulsar
        pulsar-client
        ${pulsar.version}
    
    
        org.springframework.boot
        spring-boot-starter-test
        test
    

（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 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();
    }
}

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

或

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

启动

$ 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"}'

访问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