Centos7 安装单机模式 Apache Pulsar

清疚 2022-10-07 08:54 523阅读 0赞

### Centos7 安装单机模式Apache Pulsar ###

*   *  1.1 下载Pulsar
     *  1.2 解压Pulsar
     *  1.3 进入文件夹
     *  1.4 启动单机模式Pulsar
     *  1.5 测试生产消息和消费消息
     *  1.6 退出Pulsar

## 1.1 下载Pulsar ##

wget https://archive.apache.org/dist/pulsar/pulsar-2.8.0/apache-pulsar-2.8.0-bin.tar.gz

## 1.2 解压Pulsar ##

tar xvfz apache-pulsar-2.8.0-bin.tar.gz

## 1.3 进入文件夹 ##

cd apache-pulsar-2.8.0

## 1.4 启动单机模式Pulsar ##

bin/pulsar standalone

默认启动端口是8080,如果端口被占用请修改 `/usr/local/pulsar/apache-pulsar-2.8.0/conf/standalone.conf`文件中的端口配置

## 1.5 测试生产消息和消费消息 ##

*  实例化一个叫做`first-subscription`的消费者然后订阅一个叫`my-topic`的主题

bin/pulsar-client consume my-topic -s "first-subscription"

> PS：
> 
>  *  订阅者(消费者)的名字叫做`first-subscription`，如果订阅的`topic`不存在会创建该topic
>  *  等价于 Java代码示例：
> 
>     PulsarClient client = PulsarClient.builder()
>             .serviceUrl("pulsar://localhost:6650")
>             .build();
>     Consumer consumer = client.newConsumer()
>             .topic("my-topic")
>             .subscriptionName("first-subscription")
>             .subscribe();
> 
> 如果是maven项目,pom.xml需要添加如下依赖:
> 
>     <pulsar.version>2.8.0</pulsar.version>
>      
>     <dependency>  
>        <groupId>org.apache.pulsar</groupId>  
>        <artifactId>pulsar-client</artifactId>  
>        <version>${pulsar.version}</version> 
>     </dependency>
> 
> 如果是Gradle项目，build.gradle 添加
> 
>     def pulsarVersion = '2.8.0'
>     dependencies { 
>         compile group: 'org.apache.pulsar', name: 'pulsar-client', version: pulsarVersion }

*  向my-topic 生产一条内容为hello-pulsar消息

bin/pulsar-client produce my-topic --messages "hello-pulsar"

>  *  向`my-topic`主题放入一条内容为`"hello-pulsar`" 的消息。
>  *  等价于Java代码示例：
> 
>     Producer<String> stringProducer = client.newProducer(Schema.STRING)
>            .topic("my-topic")
>            .create();
>     stringProducer.send("hello-pulsar");

## 1.6 退出Pulsar ##

如果是命令行则使用 `Ctrl+C` 终止单机模式 Pulsar 的运行。

如果是Java 代码中关闭则

producer.close();
    consumer.close();
    client.close();

或者

producer.closeAsync()
       .thenRun(() -> System.out.println("Producer closed"));
       .exceptionally((ex) -> { 
           System.err.println("Failed to close producer: " + ex);
           return ex;
       });