【mq】从零开始实现 mq-08-配置优化 fluent

【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 的配置方式，是我个人非常喜欢的一种配置方式。

传统的 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")

写起来更加丝滑流畅。

private int port = BrokerConst.DEFAULT_PORT;

private final IInvokeService invokeService = new InvokeService();

private IBrokerConsumerService registerConsumerService = new LocalBrokerConsumerService();

private IBrokerProducerService registerProducerService = new LocalBrokerProducerService();

private IMqBrokerPersist mqBrokerPersist = new LocalMqBrokerPersist();

private IBrokerPushService brokerPushService = new BrokerPushService();

private long respTimeoutMills = 5000;

private ILoadBalance loadBalance = LoadBalances.weightRoundRobbin();

private int pushMaxAttempt = 3;

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;
}

private String groupName = ProducerConst.DEFAULT_GROUP_NAME;

private String brokerAddress  = "127.0.0.1:9999";

private long respTimeoutMills = 5000;

private volatile boolean check = true;

private final IInvokeService invokeService = new InvokeService();

private final IStatusManager statusManager = new StatusManager();

private final IProducerBrokerService producerBrokerService = new ProducerBrokerService();

private long waitMillsForRemainRequest = 60 * 1000;

private ILoadBalance loadBalance = LoadBalances.weightRoundRobbin();

private int maxAttempt = 3;

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;
}

private String groupName = ConsumerConst.DEFAULT_GROUP_NAME;

private String brokerAddress  = "127.0.0.1:9999";

private long respTimeoutMills = 5000;

private volatile boolean check = true;

private long waitMillsForRemainRequest = 60 * 1000;

private final IInvokeService invokeService = new InvokeService();

private final IMqListenerService mqListenerService = new MqListenerService();

private final IStatusManager statusManager = new StatusManager();

private final IConsumerBrokerService consumerBrokerService = new ConsumerBrokerService();

private ILoadBalance loadBalance = LoadBalances.weightRoundRobbin();

private int subscribeMaxAttempt = 3;

private int unSubscribeMaxAttempt = 3;

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;
}

这一节的实现非常简单，可以说是没有啥技术难度。

只是为了让使用者更加方便。

希望本文对你有所帮助，如果喜欢，欢迎点赞收藏转发一波。

我是老马，期待与你的下次重逢。

The message queue in java.(java 简易版本 mq 实现) https://github.com/houbb/mq

rpc-从零开始实现 rpc https://github.com/houbb/rpc