作者 crossoverJie

Merge branch 'master' into cim-1.0.6

# Conflicts:
#	cim-common/src/main/java/com/crossoverjie/cim/common/data/construct/RingBufferWheel.java
#	cim-common/src/test/java/com/crossoverjie/cim/common/data/construct/RingBufferWheelTest.java
... ... @@ -26,7 +26,7 @@
- 适用于 `APP` 的消息推送中间件。
- `IOT` 海量连接场景中的消息透传中间件。
> 我有在公网部署了一套演示环境,想要体验的可以[联系我](#联系作者)加入内测群获取账号
> 在使用或开发过程中有任何疑问都可[联系我](#联系作者)
## 视频演示
... ... @@ -50,7 +50,7 @@
* [x] 路由(`cim-forward-route`)服务自身是无状态,可用 `Nginx` 代理支持高可用。
* [x] 服务端自动剔除离线客户端。
* [x] 客户端自动重连。
* [x] 延时消息
* [x] [延时消息](#延时消息)
* [ ] 分组群聊。
* [ ] SDK 开发包。
* [ ] 离线消息。
... ... @@ -183,6 +183,7 @@ java -jar cim-client-1.0.0-SNAPSHOT.jar --server.port=8084 --cim.user.id=上方
| `:pu` | 模糊匹配用户 |
| `:info` | 获取客户端信息 |
| `:emoji [option]` | 查询表情包 [option:页码] |
| `:delay [msg] [delayTime]` | 发送延时消息 |
| `:` | 更多命令正在开发中。。 |
![](https://ws3.sinaimg.cn/large/006tNbRwly1fylh7bdlo6g30go01shdt.gif)
... ... @@ -250,7 +251,15 @@ java -jar cim-client-1.0.0-SNAPSHOT.jar --server.port=8084 --cim.user.id=上方
![](https://tva1.sinaimg.cn/large/006y8mN6ly1g6j910cqrzj30dn05qjw9.jpg)
![](https://tva1.sinaimg.cn/large/006y8mN6ly1g6j99hazg6j30ax03hq35.jpg)
### 延时消息
发送 10s 的延时消息:
```shell
:delay delayMsg 10
```
![](https://tva1.sinaimg.cn/large/006y8mN6ly1g7brppmokqg30gn07gafj.gif)
## 联系作者
- [crossoverJie@gmail.com](mailto:crossoverJie@gmail.com)
... ...
... ... @@ -50,6 +50,11 @@ public final class RingBufferWheel {
*/
private volatile boolean start = false ;
/**
* total tick times
*/
private AtomicInteger tick = new AtomicInteger() ;
private Lock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
... ... @@ -115,7 +120,7 @@ public final class RingBufferWheel {
}
/**
* Start background thread to consumer wheel timer, it will run until you call method {@link #stop}
* Start background thread to consumer wheel timer, it will always run until you call method {@link #stop}
*/
public void start() {
if (!start){
... ... @@ -195,13 +200,16 @@ public final class RingBufferWheel {
}
private void size2Notify() {
try {
lock.lock();
int size = taskSize.decrementAndGet();
if (size == 0) {
condition.signal();
}
}finally {
lock.unlock();
}
}
private boolean powerOf2(int target) {
if (target < 0) {
... ... @@ -217,6 +225,7 @@ public final class RingBufferWheel {
private int mod(int target, int mod) {
// equals target % mod
target = target + tick.get() ;
return target & (mod - 1);
}
... ... @@ -273,6 +282,8 @@ public final class RingBufferWheel {
index = 0;
}
//Total tick number of records
tick.incrementAndGet();
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
... ...
package com.crossoverjie.cim.common.data.construct;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
... ... @@ -17,7 +19,7 @@ public class RingBufferWheelTest {
private static Logger logger = LoggerFactory.getLogger(RingBufferWheelTest.class) ;
public static void main(String[] args) throws InterruptedException {
test1();
test5();
return;
}
... ... @@ -25,7 +27,7 @@ public class RingBufferWheelTest {
private static void test1() throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(2) ;
Task task = new Task() ;
RingBufferWheel.Task task = new Task() ;
task.setKey(10);
RingBufferWheel wheel = new RingBufferWheel(executorService) ;
wheel.addTask(task) ;
... ... @@ -42,7 +44,7 @@ public class RingBufferWheelTest {
private static void test2() throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(2) ;
Task task = new Task() ;
RingBufferWheel.Task task = new Task() ;
task.setKey(10);
RingBufferWheel wheel = new RingBufferWheel(executorService) ;
wheel.addTask(task) ;
... ... @@ -72,7 +74,7 @@ public class RingBufferWheelTest {
private static void test3() throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(2) ;
Task task = new Task() ;
RingBufferWheel.Task task = new Task() ;
task.setKey(10);
RingBufferWheel wheel = new RingBufferWheel(executorService) ;
wheel.addTask(task) ;
... ... @@ -95,7 +97,7 @@ public class RingBufferWheelTest {
RingBufferWheel wheel = new RingBufferWheel(executorService) ;
for (int i = 0; i < 65; i++) {
Job task = new Job(i) ;
RingBufferWheel.Task task = new Job(i) ;
task.setKey(i);
wheel.addTask(task);
}
... ... @@ -114,7 +116,7 @@ public class RingBufferWheelTest {
RingBufferWheel wheel = new RingBufferWheel(executorService,512) ;
for (int i = 0; i < 65; i++) {
Job task = new Job(i) ;
RingBufferWheel.Task task = new Job(i) ;
task.setKey(i);
wheel.addTask(task);
}
... ... @@ -125,14 +127,28 @@ public class RingBufferWheelTest {
}
private static void test6() throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(2) ;
RingBufferWheel wheel = new RingBufferWheel(executorService,512) ;
private static class Task extends RingBufferWheel.Task{
@Override
public void run() {
logger.info("================");
for (int i = 0; i < 10; i++) {
RingBufferWheel.Task task = new Job(i) ;
task.setKey(i);
wheel.addTask(task);
}
TimeUnit.SECONDS.sleep(10);
RingBufferWheel.Task task = new Job(15) ;
task.setKey(15);
wheel.addTask(task);
logger.info("task size={}",wheel.taskSize());
wheel.stop(false);
}
private static void cuncrrentTest6() throws InterruptedException {
BlockingQueue<Runnable> queue = new LinkedBlockingQueue(10);
... ... @@ -184,4 +200,13 @@ public class RingBufferWheelTest {
logger.info("number={}" , num);
}
}
private static class Task extends RingBufferWheel.Task{
@Override
public void run() {
logger.info("================");
}
}
}
\ No newline at end of file
... ...