探索Pulsar之使用Go和Java操作Pulsar-阿里云开发者社区

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

2023-01-08 161

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 探索Pulsar之使用Go和Java操作Pulsar

1 前言

Pulsar官方支持的客户端库：

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

GitHub中三方的客户端库：

Go
Haskell
Scala
Rust
.NET
Node.js

具体可参看：pulsar.apache.org/docs/zh-CN/…

本次仅演示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 下载链接

github.com/apache/puls…

或

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界面：

网络异常，图片无法展示

参考：

pulsar.apache.org/docs/zh-CN/…

blog.csdn.net/ycf92124481…