add producer client

distmq-broker/src/main/java/com/github/wenweihu86/distmq/broker/BrokerMain.java

@@ -10,6 +10,7 @@
 import com.github.wenweihu86.raft.service.RaftConsensusService;
 import com.github.wenweihu86.raft.service.impl.RaftClientServiceImpl;
 import com.github.wenweihu86.raft.service.impl.RaftConsensusServiceImpl;
+import com.github.wenweihu86.raft.util.ConfigurationUtils;
 import com.github.wenweihu86.rpc.server.RPCServer;
 
 import java.util.List;
@@ -48,11 +49,21 @@ public static void main(String[] args) {
         // 注册zk
         ZKConf zkConf = conf.getZkConf();
         ZKClient zkClient = new ZKClient(zkConf);
+        zkClient.subscribeBroker();
+        zkClient.subscribeTopic();
+        // 等成为raft集群成员后，才能注册到zk
+
+        while (ConfigurationUtils.containsServer(
+                raftNode.getConfiguration(), conf.getLocalServer().getServerId())) {
+            try {
+                Thread.sleep(1000);
+            } catch (InterruptedException ex) {
+                ex.printStackTrace();
+            }
+        }
         zkClient.registerBroker(conf.getShardingId(),
                 conf.getLocalServer().getEndPoint().getHost(),
                 conf.getLocalServer().getEndPoint().getPort());
-        zkClient.subscribeBroker();
-        zkClient.subscribeTopic();
     }
 
 }

distmq-client/src/main/java/com/github/wenweihu86/distmq/client/BrokerClient.java

@@ -0,0 +1,66 @@
+package com.github.wenweihu86.distmq.client;
+
+import com.github.wenweihu86.distmq.client.api.BrokerAPI;
+import com.github.wenweihu86.rpc.client.RPCClient;
+import com.github.wenweihu86.rpc.client.RPCClientOptions;
+import com.github.wenweihu86.rpc.client.RPCProxy;
+import org.apache.commons.lang.builder.EqualsBuilder;
+import org.apache.commons.lang.builder.HashCodeBuilder;
+
+/**
+ * Created by wenweihu86 on 2017/6/24.
+ */
+public class BrokerClient {
+    private String address;
+    private RPCClient rpcClient;
+    private BrokerAPI brokerAPI;
+
+    public BrokerClient(String address, RPCClientOptions options) {
+        this.address = address;
+        this.rpcClient = new RPCClient(address, options);
+        this.brokerAPI = RPCProxy.getProxy(this.rpcClient, BrokerAPI.class);
+    }
+
+    @Override
+    public boolean equals(Object object) {
+        boolean flag = false;
+        if (object != null && BrokerClient.class.isAssignableFrom(object.getClass())) {
+            BrokerClient rhs = (BrokerClient) object;
+            flag = new EqualsBuilder()
+                    .append(address, rhs.address)
+                    .isEquals();
+        }
+        return flag;
+    }
+
+    @Override
+    public int hashCode() {
+        return new HashCodeBuilder()
+                .append(address)
+                .toHashCode();
+    }
+
+    public String getAddress() {
+        return address;
+    }
+
+    public void setAddress(String address) {
+        this.address = address;
+    }
+
+    public RPCClient getRpcClient() {
+        return rpcClient;
+    }
+
+    public void setRpcClient(RPCClient rpcClient) {
+        this.rpcClient = rpcClient;
+    }
+
+    public BrokerAPI getBrokerAPI() {
+        return brokerAPI;
+    }
+
+    public void setBrokerAPI(BrokerAPI brokerAPI) {
+        this.brokerAPI = brokerAPI;
+    }
+}

distmq-client/src/main/java/com/github/wenweihu86/distmq/client/BrokerClientManager.java

@@ -0,0 +1,39 @@
+package com.github.wenweihu86.distmq.client;
+
+import com.github.wenweihu86.rpc.client.RPCClientOptions;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * Created by wenweihu86 on 2017/6/24.
+ */
+public class BrokerClientManager {
+    private static BrokerClientManager instance;
+    private static RPCClientOptions rpcClientOptions;
+    private ConcurrentMap<String, BrokerClient> brokerClientMap = new ConcurrentHashMap<>();
+
+    public static BrokerClientManager getInstance() {
+        if (instance == null) {
+            instance = new BrokerClientManager();
+        }
+        return instance;
+    }
+
+    public static RPCClientOptions getRpcClientOptions() {
+        return rpcClientOptions;
+    }
+
+    public static void setRpcClientOptions(RPCClientOptions rpcClientOptions) {
+        BrokerClientManager.rpcClientOptions = rpcClientOptions;
+    }
+
+    public ConcurrentMap<String, BrokerClient> getBrokerClientMap() {
+        return brokerClientMap;
+    }
+
+    public void setBrokerClientMap(ConcurrentMap<String, BrokerClient> brokerClientMap) {
+        this.brokerClientMap = brokerClientMap;
+    }
+}

distmq-client/src/main/java/com/github/wenweihu86/distmq/client/CommonConfig.java

@@ -0,0 +1,45 @@
+package com.github.wenweihu86.distmq.client;
+
+import com.github.wenweihu86.distmq.client.zk.ZKConf;
+import com.github.wenweihu86.rpc.client.RPCClientOptions;
+
+/**
+ * Created by wenweihu86 on 2017/6/24.
+ */
+public class CommonConfig extends ZKConf {
+    private int brokerConnectTimeoutMs = 200;
+    private int brokerReadTimeoutMs = 500;
+    private int brokerWriteTimeoutMs = 200;
+
+    public RPCClientOptions getRPCClientOptions() {
+        RPCClientOptions rpcClientOptions = new RPCClientOptions();
+        rpcClientOptions.setConnectTimeoutMillis(brokerConnectTimeoutMs);
+        rpcClientOptions.setReadTimeoutMillis(brokerReadTimeoutMs);
+        rpcClientOptions.setWriteTimeoutMillis(brokerWriteTimeoutMs);
+        return rpcClientOptions;
+    }
+
+    public int getBrokerConnectTimeoutMs() {
+        return brokerConnectTimeoutMs;
+    }
+
+    public void setBrokerConnectTimeoutMs(int brokerConnectTimeoutMs) {
+        this.brokerConnectTimeoutMs = brokerConnectTimeoutMs;
+    }
+
+    public int getBrokerReadTimeoutMs() {
+        return brokerReadTimeoutMs;
+    }
+
+    public void setBrokerReadTimeoutMs(int brokerReadTimeoutMs) {
+        this.brokerReadTimeoutMs = brokerReadTimeoutMs;
+    }
+
+    public int getBrokerWriteTimeoutMs() {
+        return brokerWriteTimeoutMs;
+    }
+
+    public void setBrokerWriteTimeoutMs(int brokerWriteTimeoutMs) {
+        this.brokerWriteTimeoutMs = brokerWriteTimeoutMs;
+    }
+}

distmq-client/src/main/java/com/github/wenweihu86/distmq/client/consumer/ConsumerConfig.java

@@ -0,0 +1,9 @@
+package com.github.wenweihu86.distmq.client.consumer;
+
+import com.github.wenweihu86.distmq.client.CommonConfig;
+
+/**
+ * Created by wenweihu86 on 2017/6/24.
+ */
+public class ConsumerConfig extends CommonConfig {
+}

distmq-client/src/main/java/com/github/wenweihu86/distmq/client/producer/Producer.java

@@ -0,0 +1,100 @@
+package com.github.wenweihu86.distmq.client.producer;
+
+import com.github.wenweihu86.distmq.client.BrokerClient;
+import com.github.wenweihu86.distmq.client.BrokerClientManager;
+import com.github.wenweihu86.distmq.client.api.BrokerMessage;
+import com.github.wenweihu86.distmq.client.zk.ZKClient;
+import com.github.wenweihu86.distmq.client.zk.ZKData;
+import com.google.protobuf.ByteString;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * Created by wenweihu86 on 2017/6/24.
+ */
+public class Producer {
+    private static final Logger LOG = LoggerFactory.getLogger(Producer.class);
+    private ProducerConfig config;
+    private ZKClient zkClient;
+
+    public Producer(ProducerConfig config) {
+        this.config = config;
+        BrokerClientManager.setRpcClientOptions(this.config.getRPCClientOptions());
+        zkClient = new ZKClient(config);
+        zkClient.subscribeBroker();
+        zkClient.subscribeTopic();
+    }
+
+    public boolean send(String topic, byte[] messageBytes) {
+        ZKData zkData = ZKData.getInstance();
+        Map<Integer, Integer> queueMap;
+        zkData.getTopicLock().lock();
+        try {
+            queueMap = zkData.getTopicMap().get(topic);
+        } finally {
+            zkData.getTopicLock().unlock();
+        }
+        Integer queueId;
+        Integer shardingId;
+        if (queueMap == null) {
+            zkClient.registerTopic(topic, config.getQueueCountPerTopic());
+            zkData.getTopicLock().lock();
+            try {
+                while (!zkData.getTopicMap().containsKey(topic)) {
+                    zkData.getTopicCondition().awaitUninterruptibly();
+                }
+            } finally {
+                zkData.getTopicLock().unlock();
+            }
+        }
+
+        zkData.getTopicLock().lock();
+        try {
+            queueMap = zkData.getTopicMap().get(topic);
+            int queueCount = queueMap.size();
+            int randomIndex = ThreadLocalRandom.current().nextInt(0, queueCount);
+            Integer[] queueArray = queueMap.keySet().toArray(new Integer[0]);
+            queueId = queueArray[randomIndex];
+            shardingId = queueMap.get(queueId);
+        } finally {
+            zkData.getTopicLock().unlock();
+        }
+
+        // send message to broker
+        BrokerMessage.SendMessageRequest request = BrokerMessage.SendMessageRequest.newBuilder()
+                .setTopic(topic)
+                .setQueue(queueId)
+                .setContent(ByteString.copyFrom(messageBytes))
+                .build();
+
+        List<String> brokerAddressList;
+        zkData.getBrokerLock().lock();
+        try {
+            brokerAddressList = zkData.getBrokerMap().get(shardingId);
+        } finally {
+            zkData.getBrokerLock().unlock();
+        }
+        int randIndex = ThreadLocalRandom.current().nextInt(0, brokerAddressList.size());
+        String brokerAddress = brokerAddressList.get(randIndex);
+        BrokerClient brokerClient = BrokerClientManager.getInstance().getBrokerClientMap().get(brokerAddress);
+        BrokerMessage.SendMessageResponse response = brokerClient.getBrokerAPI().sendMessage(request);
+        if (response == null || response.getBaseRes().getResCode() != BrokerMessage.ResCode.RES_CODE_SUCCESS) {
+            LOG.warn("send message failed, topic={}, queue={}, brokerAddress={}",
+                    topic, queueId, brokerAddress);
+            return false;
+        }
+        return true;
+    }
+
+    public ProducerConfig getConfig() {
+        return config;
+    }
+
+    public void setConfig(ProducerConfig config) {
+        this.config = config;
+    }
+}

distmq-client/src/main/java/com/github/wenweihu86/distmq/client/producer/ProducerConfig.java

@@ -0,0 +1,18 @@
+package com.github.wenweihu86.distmq.client.producer;
+
+import com.github.wenweihu86.distmq.client.CommonConfig;
+
+/**
+ * Created by wenweihu86 on 2017/6/24.
+ */
+public class ProducerConfig extends CommonConfig {
+    private int queueCountPerTopic = 4;
+
+    public int getQueueCountPerTopic() {
+        return queueCountPerTopic;
+    }
+
+    public void setQueueCountPerTopic(int queueCountPerTopic) {
+        this.queueCountPerTopic = queueCountPerTopic;
+    }
+}