# cgyun-tiny-pulsar

**Repository Path**: climbing-the-mountain/cgyun-tiny-pulsar

## Basic Information

- **Project Name**: cgyun-tiny-pulsar
- **Description**: Pulsar是一个用于服务端到服务端的消息中间件，具有多租户、高性能等优势。Pulsar最初由Yahoo开发，目前由Apache软件基金会管理。Pulsar采用发布-订阅的设计模式，Producer发布消息到Topic，Consumer订阅Topic、处理Topic中的消息。
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 1
- **Created**: 2021-08-14
- **Last Updated**: 2023-07-13

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# cgyun-tiny-pulsar

#### 介绍
Pulsar是一个用于服务端到服务端的消息中间件，具有多租户、高性能等优势。Pulsar最初由Yahoo开发，目前由Apache软件基金会管理。Pulsar采用发布-订阅的设计模式，Producer发布消息到Topic，Consumer订阅Topic、处理Topic中的消息。

#### Pulsar具有如下特性
- Pulsar的单个实例原生支持集群。
- 极低的发布延迟和端到端延迟。
- 可无缝扩展到超过一百万个Topic。
- 简单易用的客户端API，支持Java、Go、Python和C++。
- 支持多种Topic订阅模式（独占订阅、共享订阅、故障转移订阅）。
- 通过Apache BookKeeper提供的持久化消息存储机制保证消息传递。



#### Pulsar架构

![4](./img/image.png)

#### Pulsar安装

我们选择通过docker安装

```shell
docker run -it -p 6650:6650  -p 8091:8080  \ 
--mount source=pulsardata,target=/data/pulsar/data \ 
--mount source=pulsarconf,target=/data/pulsar/conf \ 
-e advertisedAddress=pulsar-standalone \ 
--name pulscar-standalone \ 
--hostname pulsar-standalone \ apachepulsar/pulsar:2.8.0  \ 
sh -c "bin/apply-config-from-env.py conf/standalone.conf&&bin/pulsar standalone -nfw -nss"
```

下载完成后运行pulsar-manager容器，从9527端口可以访问Web页面；

```shell
docker run -d -it --name pulsar-manager\    
-p 9527:9527 -p 7750:7750 \    
-e SPRING_CONFIGURATION_FILE=/pulsar-manager/pulsar-manager/application.properties \    
--link pulsar-standalone \     
apachepulsar/pulsar-manager:v0.2.0
```

![4](./img/1.png)

运行成功后，我们刚开始无法访问，需要创建管理员账号，这里创建账号为admin:apachepulsar：

```shell
CSRF_TOKEN=$(curl http://localhost: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://localhost:7750/pulsar-manager/users/superuser \    
-d '{"name": "admin", "password": "apachepulsar", "description": "test", "email": "username@test.org"}'
```
输入用户名字密码，进去可视化管理后台：
![4](./img/2.png)
可以创建消息主题
![4](./img/3.png)

一般在生成环境当中，我们不会手动创建消息主题，一般都会结合项目自动创建，接下来我们测试下Pulsar结合SpringBoot的使用

####  Pulsar结合SpringBoot使用

>  Pulsar结合SpringBoot使用也是非常简单的，我们可以使用Pulsar官方的Java SDK，也可以使用第三方的SpringBoot Starter。这里使用Starter，非常简单！

 首先在`pom.xml`中添加Pulsar相关依赖；

```xml
<!--SpringBoot整合Pulsar-->
<dependency>
    <groupId>io.github.majusko</groupId>
    <artifactId>pulsar-java-spring-boot-starter</artifactId>
    <version>1.0.4</version>
</dependency>
```

然后在`application.yml`中添加Pulsar的`Service URL`配置

```properties
server:
  port: 8088
pulsar:
  service-url: pulsar://121.xxx.212.xx:6650
```

再添加Pulsar的Java配置，声明两个Topic，并确定好发送的消息类型；

```java
@Configuration
public class PulsarConfig {
    @Bean
    public ProducerFactory producerFactory() {
        return new ProducerFactory()
                .addProducer("bootTopic", MessageDto.class)
                .addProducer("stringTopic", String.class);
    }
}
```

创建Pulsar生产者，往Topic中发送消息，这里可以发现Pulsar是支持直接发送消息对象的；

 ```java
  @Component
  public class PulsarProducer {
      @Autowired
      private PulsarTemplate<MessageDto> template;
  
      public void send(MessageDto message){
          try {
              template.send("bootTopic",message);
          } catch (PulsarClientException e) {
              e.printStackTrace();
          }
      }
  }
 ```

创建Pulsar消费者，从Topic中获取并消费消息，也是可以直接获取到消息对象的；

```java
@Slf4j
@Component
public class PulsarRealConsumer {

    @PulsarConsumer(topic="bootTopic", clazz= MessageDto.class)
    public void consume(MessageDto message) {
        log.info("PulsarRealConsumer consume id:{},content:{}",message.getId(),message.getContent());
    }
}
```

添加测试接口，调用生产者发送消息；

```java
@Api(tags = "PulsarController", description = "Pulsar功能测试")
@Controller
@RequestMapping("/pulsar")
public class PulsarController {

    @Autowired
    private PulsarProducer pulsarProducer;

    @ApiOperation("发送消息")
    @RequestMapping(value = "/sendMessage", method = RequestMethod.POST)
    @ResponseBody
    public CommonResult sendMessage(@RequestBody MessageDto message) {
        pulsarProducer.send(message);
        return CommonResult.success(null);
    }
}
```



启动sringboot项目，会自动连接pulsar，成功后会并且打印配置信息，如果失败则项目会自动停止。


![4](./img/5.png)

此时我们也可以通过管理后台查看注册的主题和当前的消费者信息；

![4](./img/6.png)
![4](./img/8.png)
在Swagger中调用接口进行测试；

![4](./img/4.png)
调用成功后，控制台将输入如下信息，表示消息已经被成功接收并消费了。

```powershell
2021-08-14 16:49:53.740  INFO 8169: 
PulsarRealConsumer consume id:0,content:我是panda
```

#### 总结

Pulsar对Docker支持无疑是更好的，官方文档也更全。对比下图形化工具`Pulsar Manager`和`Kafka Eagle`，Pulsar的图形化工具感觉有点简陋。介于目前雅虎、腾讯、360等互联网大厂都在使用Pulsar，Pulsar的性能和稳定性应该是很不错的！

#### 码云链接

[cgyun-tiny-pulsar](https://gitee.com/climbing-the-mountain/cgyun-tiny-pulsar)


#### 知乎博客
![](https://pic3.zhimg.com/v2-c1c8af5273694911db3f6062d0d25ab2_1440w.jpg?source=172ae18b)

[springboot+消息中间件pulsar，可视化管理+文档贼全，准备起飞](https://zhuanlan.zhihu.com/p/399788487)

#### 参与贡献

Pulsar官方文档：https://pulsar.apache.org/docs/en/standalone-docker/
SpringBoot Starter官方文档：https://github.com/majusko/pulsar-java-spring-boot-starter
再见RocketMQ！全新一代消息中间件，带可视化管理，文档贼全！ https://blog.csdn.net/zhenghongcs/article/details/118927256Apache 

Pulsar 上手实战第二期：容器部署实战 https://www.bilibili.com/video/BV1nq4y1H7pW?from=search&seid=9047391456949428020


#### 特技

1.  使用 Readme\_XXX.md 来支持不同的语言，例如 Readme\_en.md, Readme\_zh.md
2.  Gitee 官方博客 [blog.gitee.com](https://blog.gitee.com)
3.  你可以 [https://gitee.com/explore](https://gitee.com/explore) 这个地址来了解 Gitee 上的优秀开源项目
4.  [GVP](https://gitee.com/gvp) 全称是 Gitee 最有价值开源项目，是综合评定出的优秀开源项目
5.  Gitee 官方提供的使用手册 [https://gitee.com/help](https://gitee.com/help)
6.  Gitee 封面人物是一档用来展示 Gitee 会员风采的栏目 [https://gitee.com/gitee-stars/](https://gitee.com/gitee-stars/)