- 博客(17)
- 收藏
- 关注
RSS订阅
原创 Pulsar源码解析-延迟队列实现
问题：延迟队列的作用延迟队列数据结构如何添加到延迟队列?如何从延迟队列取出?问题1延迟队列的作用延迟队列用来解决需要延迟消费的场景，例如 电商中订单超时15分钟未支付自动关闭特点：Pulsar的延迟队列可支持任意维度的延迟存储：堆外内存消息数：1G堆外内存 = 1 * 1024 * 1024 * 1024 / 24 = 44,739,242条，1G支持4400万条可满足大多数业务场景，不足就加内存。（ps: 消息是持久化到磁盘，延迟消息是从磁盘读取出来放入延迟队列中，只放消息的索引：2
2022-01-06 19:31:50
2680
原创 Pulsar源码解析
一、消息的生产与发送连接建立Pulsar源码解析-客户端-PulsarClient构建生产者创建(包含Topic创建原理)Pulsar源码解析-非分区Producer创建的底层实现Pulsar源码解析-分区Producer创建的底层实现消费者创建Pulsar源码解析-客户端-单个消费者ConsumerImpl创建底层实现Pulsar源码解析-客户端-多消费者MultiTopicsConsumerImpl创建底层实现Pulsar源码解析-服务端-消费者创建的底层实现生产者-发送消息原理
2021-12-30 17:35:51
2973
原创 Pulsar源码解析-客户端-消费者接收消息的底层实现
上一篇介绍了消费者请求拉取消息，最后发送给消费者。本章介绍消费者接收。一、客户端消息接收入口public class ClientCnx { protected void handleMessage(CommandMessage cmdMessage, ByteBuf headersAndPayload) { checkArgument(state == State.Ready); ConsumerImpl<?> consumer = consu.
2021-12-30 17:02:08
1114
原创 Pulsar源码解析-服务端-消费者消息拉取的底层实现
上一篇介绍了服务端消费者的创建，本章介绍消息拉取请求实现一、消息拉取请求入口public class ServerCnx { protected void handleFlow(CommandFlow flow) { checkArgument(state == State.Connected); CompletableFuture<Consumer> consumerFuture = consumers.get(flow.getConsume.
2021-12-30 00:39:08
2076
原创 Pulsar源码解析-服务端-消费者创建的底层实现
上一篇介绍的是消费者客户端的创建，本章解析服务端创建一、消费者服务端创建入口由于代码量较多，省略非重点public class ServerCnx {protected void handleSubscribe(final CommandSubscribe subscribe) { // 参数含义客户端消费者构造中都介绍过 final long requestId = subscribe.getRequestId(); final long consumer.
2021-12-28 18:40:47
1198
原创 Pulsar源码解析-客户端-多消费者MultiTopicsConsumerImpl创建底层实现
上一篇介绍了单个消费者的创建实现，本章介绍多个消费者一、MultiTopicsConsumerImpl创建入口public class PulsarClientImpl {public <T> CompletableFuture<Consumer<T>> subscribeAsync(ConsumerConfigurationData<T> conf, Schema<T> schema, ConsumerInterceptors<.
2021-12-28 14:00:44
1449
2
原创 Pulsar源码解析-客户端-单个消费者ConsumerImpl创建底层实现
pulsar消费者有多种ConsumerImpl：一个消费者客户端连接MultiTopicsConsumerImpl：多个消费者，topic配置有多个，如果开启重试队列也是当前实现，具体看MultiTopicsConsumerImpl源码解析PatternMultiTopicsConsumerImpl：配置正则表达式的topic，本质也是MultiTopicsConsumerImpl，主要实现了topic的自动发现。ZeroQueueConsumerImpl：没有缓冲队列的消费者 即receive.
2021-12-27 22:57:15
1222
原创 Pulsar源码解析-服务端-Producer消息发送底层实现
之前介绍了客户端的发送实现，客户端发送到服务端，服务端怎么处理消息的我们看一下源码实现一、发送消息服务端入口public class ServerCnx { protected void handleSend(CommandSend send, ByteBuf headersAndPayload) { // 获取创建时存储的生产者 CompletableFuture<Producer> producerFuture = producers.get(sen.
2021-12-23 23:36:17
1633
5
原创 Pulsar源码解析-客户端-生产者消息发送底层实现
前几篇介绍过生产者创建的底层实现，本章介绍发送的实现细节一、分区生产者和非分区生产发送的差异消息发送存在分区和非分区的Producer代码实现，先介绍一下区别点，然后后面流程都一样。分区Producer需要实现选择一个具体的Producer实现，怎么选择的呢？创建生产者时会配置路由策略，看一下实现：public class PartitionedProducerImpl<T> { // 调用是在PartitionedProducerImpl构造中 private Mess.
2021-12-22 22:22:23
1150
原创 Pulsar源码解析-分区Producer创建的底层实现
上一篇介绍过Pulsar源码解析-非分区Topic的Producer底层实现原理差别不大，主要在客户端，回到创建Producer时的代码 private <T> CompletableFuture<Producer<T>> createProducerAsync(String topic, ProducerConfigurati
2021-12-22 19:04:17
886
原创 Pulsar源码解析-非分区Producer创建的底层实现
本章将详细介绍Producer创建底层的实现原理，介绍过程的代码是有删减的，只保留最核心的那部分，像参数校验这些会直接舍去，防止视觉干扰一、客户端Producer创建入口Producer创建是调用了PulsarClient#newProducerpublic class PulsarClientImpl implements PulsarClient { public ProducerBuilder<byte[]> newProducer() { return.
2021-12-22 17:19:11
1511
原创 Pulsar源码解析-客户端-PulsarClient构建
看源码最好理解的就是从使用开始，本节从客户端连接构建开始，逐步到broker一、使用案例PulsarClient build = PulsarClient.builder() .listenerThreads(1) .ioThreads(1) .serviceUrl("pulsar://127.0.0.1:6650") .build();这是建造者模式，参数比较多使用建.
2021-12-21 20:15:29
1776
原创 Mac安装Prometheus Grafana
查找prometheus.yml，通过上面查看安装路径复制前缀查询。(mac不同系统版本有差异，所以要搜索出来)修改prometheus.yml新增个任务。默认prometheus.yml。默认密码admin/admin。启动prometheus。
2022-08-02 16:17:12
351
原创 简单好用pulsar spring boot starter
pulsar spring boot starter简介https://github.com/mingyifei/pulsar-spring-boot-starter普遍性：同一个Topic支持多种任意数据类型传输 例如： 第一个发送A对象，消费为A。第二次发送B对象，接收也可以是A，只要字段对应上即可。 也可以Map接收，数据类型无限制，发什么都可以。所有配置全面支持SpringEL表达式配置通过PulsarProducer注解enableChunking开启配置支持大型消息传输支持超
2021-12-14 14:20:02
1351
空空如也
空空如也
TA创建的收藏夹 TA关注的收藏夹
TA关注的人