-
Notifications
You must be signed in to change notification settings - Fork 9
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
859572d
commit 11b5aa9
Showing
9 changed files
with
265 additions
and
2 deletions.
There are no files selected for viewing
1 change: 1 addition & 0 deletions
1
distmq-broker/src/main/java/com/github/wenweihu86/distmq/broker/BrokerMain.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
1 change: 1 addition & 0 deletions
1
distmq-broker/src/main/java/com/github/wenweihu86/distmq/broker/BrokerStateMachine.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
distmq-broker/src/main/java/com/github/wenweihu86/distmq/broker/log/SegmentedLog.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
125 changes: 125 additions & 0 deletions
125
distmq-client/src/main/java/com/github/wenweihu86/distmq/client/zk/ZKClient.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,125 @@ | ||
| package com.github.wenweihu86.distmq.client.zk; | ||
|
|
||
| import org.apache.curator.RetryPolicy; | ||
| import org.apache.curator.framework.CuratorFramework; | ||
| import org.apache.curator.framework.CuratorFrameworkFactory; | ||
| import org.apache.curator.framework.api.CuratorWatcher; | ||
| import org.apache.curator.retry.ExponentialBackoffRetry; | ||
| import org.apache.zookeeper.CreateMode; | ||
| import org.apache.zookeeper.WatchedEvent; | ||
| import org.apache.zookeeper.Watcher; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.util.Iterator; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
|
|
||
| /** | ||
| * Created by wenweihu86 on 2017/6/21. | ||
| */ | ||
| public class ZKClient { | ||
| private static final Logger LOG = LoggerFactory.getLogger(ZKClient.class); | ||
| private ZKConf zkConf; | ||
| private CuratorFramework zkClient; | ||
|
|
||
| public ZKClient(ZKConf conf) { | ||
| this.zkConf = conf; | ||
| RetryPolicy retryPolicy = new ExponentialBackoffRetry( | ||
| zkConf.getRetryIntervalMs(), zkConf.getRetryCount()); | ||
| this.zkClient = CuratorFrameworkFactory.builder() | ||
| .connectString(zkConf.getServers()) | ||
| .retryPolicy(retryPolicy) | ||
| .connectionTimeoutMs(zkConf.getConnectTimeoutMs()) | ||
| .sessionTimeoutMs(zkConf.getSessionTimeoutMs()) | ||
| .build(); | ||
| this.zkClient.start(); | ||
| } | ||
|
|
||
| public void registerBroker(int shardingId, String ip, int port) { | ||
| String path = String.format("%s/brokers/%d/%s:%d", | ||
| zkConf.getBasePath(), shardingId, ip, port); | ||
| try { | ||
| zkClient.create() | ||
| .creatingParentsIfNeeded() | ||
| .withMode(CreateMode.EPHEMERAL) | ||
| .forPath(path, "".getBytes()); | ||
| } catch (Exception ex) { | ||
| LOG.warn("registerBroker exception:", ex); | ||
| } | ||
| } | ||
|
|
||
| public void subscribeBroker() { | ||
| final ZKData zkData = ZKData.getInstance(); | ||
| final Map<Integer, List<String>> brokerMap = zkData.getBrokerMap(); | ||
| try { | ||
| final String brokerParentPath = zkConf.getBasePath() + "/brokers"; | ||
| List<String> shardings = zkClient.getChildren().forPath(brokerParentPath); | ||
| for (String sharding : shardings) { | ||
| final int shardingId = Integer.valueOf(sharding); | ||
| final String shardingPath = brokerParentPath + "/" + sharding; | ||
| List<String> brokerAddressList = zkClient.getChildren().forPath(shardingPath); | ||
| brokerMap.put(shardingId, brokerAddressList); | ||
| zkClient.getChildren().usingWatcher( | ||
| new BrokerShardingChildrenWather(shardingId)) | ||
| .forPath(shardingPath); | ||
| } | ||
| zkClient.getChildren().usingWatcher(new BrokerChildrenWatcher()).forPath(brokerParentPath); | ||
| } catch (Exception ex) { | ||
| LOG.warn("subscribeBroker exception:", ex); | ||
| } | ||
| } | ||
|
|
||
| public void createTopic(String topic, int queueNum) { | ||
| } | ||
|
|
||
| private class BrokerShardingChildrenWather implements CuratorWatcher { | ||
| private int shardingId; | ||
|
|
||
| public BrokerShardingChildrenWather(int shardingId) { | ||
| this.shardingId = shardingId; | ||
| } | ||
|
|
||
| @Override | ||
| public void process(WatchedEvent event) throws Exception { | ||
| ZKData zkData = ZKData.getInstance(); | ||
| Map<Integer, List<String>> brokerMap = zkData.getBrokerMap(); | ||
| if (event.getType() == Watcher.Event.EventType.NodeChildrenChanged) { | ||
| String shardingPath = zkConf.getBasePath() + "/brokers/" + shardingId; | ||
| List<String> newBrokerAddressList = zkClient.getChildren().forPath(shardingPath); | ||
| // TODO: 对于client需要关闭被删除节点的连接，以及新建新增节点连接 | ||
| brokerMap.put(shardingId, newBrokerAddressList); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private class BrokerChildrenWatcher implements CuratorWatcher { | ||
| @Override | ||
| public void process(WatchedEvent event) throws Exception { | ||
| ZKData zkData = ZKData.getInstance(); | ||
| Map<Integer, List<String>> brokerMap = zkData.getBrokerMap(); | ||
| if (event.getType() == Watcher.Event.EventType.NodeChildrenChanged) { | ||
| String brokerPath = zkConf.getBasePath() + "/brokers"; | ||
| List<String> newShardings = zkClient.getChildren().forPath(brokerPath); | ||
| Iterator<Map.Entry<Integer, List<String>>> iterator = brokerMap.entrySet().iterator(); | ||
| while (iterator.hasNext()){ | ||
| Map.Entry<Integer, List<String>> entry = iterator.next(); | ||
| if (!newShardings.contains(Integer.valueOf(entry.getKey()))) { | ||
| // TODO:对于client，需要删除对应节点的连接 | ||
| iterator.remove(); | ||
| } | ||
| } | ||
| for (String sharding : newShardings) { | ||
| int shardingId = Integer.valueOf(sharding); | ||
| if (!brokerMap.containsKey(shardingId)) { | ||
| String shardingPath = brokerPath + "/" + sharding; | ||
| zkClient.getChildren().usingWatcher( | ||
| new BrokerShardingChildrenWather(shardingId)) | ||
| .forPath(shardingPath); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } |
61 changes: 61 additions & 0 deletions
61
distmq-client/src/main/java/com/github/wenweihu86/distmq/client/zk/ZKConf.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| package com.github.wenweihu86.distmq.client.zk; | ||
|
|
||
| /** | ||
| * Created by wenweihu86 on 2017/6/21. | ||
| */ | ||
| public class ZKConf { | ||
| private String servers; | ||
| private int connectTimeoutMs; | ||
| private int sessionTimeoutMs; | ||
| private int retryCount; | ||
| private int retryIntervalMs; | ||
| private String basePath; | ||
|
|
||
| public String getServers() { | ||
| return servers; | ||
| } | ||
|
|
||
| public void setServers(String servers) { | ||
| this.servers = servers; | ||
| } | ||
|
|
||
| public int getConnectTimeoutMs() { | ||
| return connectTimeoutMs; | ||
| } | ||
|
|
||
| public void setConnectTimeoutMs(int connectTimeoutMs) { | ||
| this.connectTimeoutMs = connectTimeoutMs; | ||
| } | ||
|
|
||
| public int getSessionTimeoutMs() { | ||
| return sessionTimeoutMs; | ||
| } | ||
|
|
||
| public void setSessionTimeoutMs(int sessionTimeoutMs) { | ||
| this.sessionTimeoutMs = sessionTimeoutMs; | ||
| } | ||
|
|
||
| public int getRetryCount() { | ||
| return retryCount; | ||
| } | ||
|
|
||
| public void setRetryCount(int retryCount) { | ||
| this.retryCount = retryCount; | ||
| } | ||
|
|
||
| public int getRetryIntervalMs() { | ||
| return retryIntervalMs; | ||
| } | ||
|
|
||
| public void setRetryIntervalMs(int retryIntervalMs) { | ||
| this.retryIntervalMs = retryIntervalMs; | ||
| } | ||
|
|
||
| public String getBasePath() { | ||
| return basePath; | ||
| } | ||
|
|
||
| public void setBasePath(String basePath) { | ||
| this.basePath = basePath; | ||
| } | ||
| } |
34 changes: 34 additions & 0 deletions
34
distmq-client/src/main/java/com/github/wenweihu86/distmq/client/zk/ZKData.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| package com.github.wenweihu86.distmq.client.zk; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
|
|
||
| /** | ||
| * Created by huwenwei on 2017/6/21. | ||
| */ | ||
| public class ZKData { | ||
| private static ZKData instance; | ||
|
|
||
| public static ZKData getInstance() { | ||
| if (instance == null) { | ||
| instance = new ZKData(); | ||
| } | ||
| return instance; | ||
| } | ||
|
|
||
| // shardingId -> broker address list | ||
| private Map<Integer, List<String>> brokerMap = new ConcurrentHashMap<>(); | ||
|
|
||
| public static void setInstance(ZKData instance) { | ||
| ZKData.instance = instance; | ||
| } | ||
|
|
||
| public Map<Integer, List<String>> getBrokerMap() { | ||
| return brokerMap; | ||
| } | ||
|
|
||
| public void setBrokerMap(Map<Integer, List<String>> brokerMap) { | ||
| this.brokerMap = brokerMap; | ||
| } | ||
| } |