前景回顾
【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
fluent
大家好，我是老马。
fluent 的配置形式，是我集体十分喜爱的一种配置形式。
传统的 java 应用 get/set 办法进行属性设置。
相似这种：
MqBroker mqBroker = 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<ConsumerSubscribeBo> 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<ConsumerSubscribeBo> 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<RpcChannelFuture> 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<RpcChannelFuture> 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<RpcChannelFuture> 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<RpcChannelFuture> loadBalance) { this.loadBalance = loadBalance; return this;}小结
这一节的实现非常简单，能够说是没有啥技术难度。
只是为了让使用者更加不便。
心愿本文对你有所帮忙，如果喜爱，欢送点赞珍藏转发一波。
我是老马，期待与你的下次重逢。
开源地址
The message queue in java.(java 繁难版本 mq 实现) https://github.com/houbb/mq
拓展浏览
rpc-从零开始实现 rpc https://github.com/houbb/rpc