【mq】从零开始实现 mq-08-配置优化 fluent mq消息队列开源手写组件

前景回顾【mq】从零开始实现 mq-01-生产者、消费者启动
【mq】从零开始实现 mq-02-如何实现生产者调用消费者？
【mq】从零开始实现 mq-03-引入 broker 中间人
【mq】从零开始实现 mq-04-启动检测与实现优化
【mq】从零开始实现 mq-05-实现优雅停机
【mq】从零开始实现 mq-06-消费者心跳检测 heartbeat
【mq】从零开始实现 mq-07-负载均衡 load balance
【【mq】从零开始实现 mq-08-配置优化 fluent】【mq】从零开始实现 mq-08-配置优化 fluent
fluent 大家好，我是老马。
fluent 的配置方式，是我个人非常喜欢的一种配置方式。
传统的 java 使用 get/set 方法进行属性设置。
类似这种：

MqBrokermqBroker = new MqBroker(); mqBroker.setPort(9999); mqBroker.setAddress("127.0.0.1");

fluent 写法可以让我们写起来代码更加流畅：

MqBroker.newInstance() .port(9999) .address("127.0.0.1")

写起来更加丝滑流畅。

文章图片

Broker 配置属性

/** * 端口号 */ private int port = BrokerConst.DEFAULT_PORT; /** * 调用管理类 * * @since 1.0.0 */ private final IInvokeService invokeService = new InvokeService(); /** * 消费者管理 * * @since 0.0.3 */ private IBrokerConsumerService registerConsumerService = new LocalBrokerConsumerService(); /** * 生产者管理 * * @since 0.0.3 */ private IBrokerProducerService registerProducerService = new LocalBrokerProducerService(); /** * 持久化类 * * @since 0.0.3 */ private IMqBrokerPersist mqBrokerPersist = new LocalMqBrokerPersist(); /** * 推送服务 * * @since 0.0.3 */ private IBrokerPushService brokerPushService = new BrokerPushService(); /** * 获取响应超时时间 * @since 0.0.3 */ private long respTimeoutMills = 5000; /** * 负载均衡 * @since 0.0.7 */ private ILoadBalance loadBalance = LoadBalances.weightRoundRobbin(); /** * 推送最大尝试次数 * @since 0.0.8 */ private int pushMaxAttempt = 3;

flent 配置

public MqBroker port(int port) { this.port = port; return this; }public MqBroker registerConsumerService(IBrokerConsumerService registerConsumerService) { this.registerConsumerService = registerConsumerService; return this; }public MqBroker registerProducerService(IBrokerProducerService registerProducerService) { this.registerProducerService = registerProducerService; return this; }public MqBroker mqBrokerPersist(IMqBrokerPersist mqBrokerPersist) { this.mqBrokerPersist = mqBrokerPersist; return this; }public MqBroker brokerPushService(IBrokerPushService brokerPushService) { this.brokerPushService = brokerPushService; return this; }public MqBroker respTimeoutMills(long respTimeoutMills) { this.respTimeoutMills = respTimeoutMills; return this; }public MqBroker loadBalance(ILoadBalance loadBalance) { this.loadBalance = loadBalance; return this; }

Producer 配置属性

/** * 分组名称 */ private String groupName = ProducerConst.DEFAULT_GROUP_NAME; /** * 中间人地址 */ private String brokerAddress= "127.0.0.1:9999"; /** * 获取响应超时时间 * @since 0.0.2 */ private long respTimeoutMills = 5000; /** * 检测 broker 可用性 * @since 0.0.4 */ private volatile boolean check = true; /** * 调用管理服务 * @since 0.0.2 */ private final IInvokeService invokeService = new InvokeService(); /** * 状态管理类 * @since 0.0.5 */ private final IStatusManager statusManager = new StatusManager(); /** * 生产者-中间服务端服务类 * @since 0.0.5 */ private final IProducerBrokerService producerBrokerService = new ProducerBrokerService(); /** * 为剩余的请求等待时间 * @since 0.0.5 */ private long waitMillsForRemainRequest = 60 * 1000; /** * 负载均衡策略 * @since 0.0.7 */ private ILoadBalance loadBalance = LoadBalances.weightRoundRobbin(); /** * 消息发送最大尝试次数 * @since 0.0.8 */ private int maxAttempt = 3;

fluent 配置

public MqProducer groupName(String groupName) { this.groupName = groupName; return this; }public MqProducer brokerAddress(String brokerAddress) { this.brokerAddress = brokerAddress; return this; }public MqProducer respTimeoutMills(long respTimeoutMills) { this.respTimeoutMills = respTimeoutMills; return this; }public MqProducer check(boolean check) { this.check = check; return this; }public MqProducer waitMillsForRemainRequest(long waitMillsForRemainRequest) { this.waitMillsForRemainRequest = waitMillsForRemainRequest; return this; }public MqProducer loadBalance(ILoadBalance loadBalance) { this.loadBalance = loadBalance; return this; }public MqProducer maxAttempt(int maxAttempt) { this.maxAttempt = maxAttempt; return this; }

Consuemr 配置属性

/** * 组名称 */ private String groupName = ConsumerConst.DEFAULT_GROUP_NAME; /** * 中间人地址 */ private String brokerAddress= "127.0.0.1:9999"; /** * 获取响应超时时间 * @since 0.0.2 */ private long respTimeoutMills = 5000; /** * 检测 broker 可用性 * @since 0.0.4 */ private volatile boolean check = true; /** * 为剩余的请求等待时间 * @since 0.0.5 */ private long waitMillsForRemainRequest = 60 * 1000; /** * 调用管理类 * * @since 1.0.0 */ private final IInvokeService invokeService = new InvokeService(); /** * 消息监听服务类 * @since 0.0.5 */ private final IMqListenerService mqListenerService = new MqListenerService(); /** * 状态管理类 * @since 0.0.5 */ private final IStatusManager statusManager = new StatusManager(); /** * 生产者-中间服务端服务类 * @since 0.0.5 */ private final IConsumerBrokerService consumerBrokerService = new ConsumerBrokerService(); /** * 负载均衡策略 * @since 0.0.7 */ private ILoadBalance loadBalance = LoadBalances.weightRoundRobbin(); /** * 订阅最大尝试次数 * @since 0.0.8 */ private int subscribeMaxAttempt = 3; /** * 取消订阅最大尝试次数 * @since 0.0.8 */ private int unSubscribeMaxAttempt = 3;

fluent 配置

public MqConsumerPush subscribeMaxAttempt(int subscribeMaxAttempt) { this.subscribeMaxAttempt = subscribeMaxAttempt; return this; }public MqConsumerPush unSubscribeMaxAttempt(int unSubscribeMaxAttempt) { this.unSubscribeMaxAttempt = unSubscribeMaxAttempt; return this; }public MqConsumerPush groupName(String groupName) { this.groupName = groupName; return this; }public MqConsumerPush brokerAddress(String brokerAddress) { this.brokerAddress = brokerAddress; return this; }public MqConsumerPush respTimeoutMills(long respTimeoutMills) { this.respTimeoutMills = respTimeoutMills; return this; }public MqConsumerPush check(boolean check) { this.check = check; return this; }public MqConsumerPush waitMillsForRemainRequest(long waitMillsForRemainRequest) { this.waitMillsForRemainRequest = waitMillsForRemainRequest; return this; }public MqConsumerPush loadBalance(ILoadBalance loadBalance) { this.loadBalance = loadBalance; return this; }

小结这一节的实现非常简单，可以说是没有啥技术难度。
只是为了让使用者更加方便。
希望本文对你有所帮助，如果喜欢，欢迎点赞收藏转发一波。
我是老马，期待与你的下次重逢。
开源地址