用redis来贯彻全体ack机制的音讯队列

那样大家就应用Redis达成了八个近乎于Java
的原子类的效果。在实际上的Web开发中,我们能够采纳redis来解决能源重复修改或争用的题材。

贯彻思路

package com.redis.lock;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.Transaction;
/**
 * topic:利用redis的事务,实现一个乐观锁
 * 
 * @author zhiming
 *
 */
public class RedisWatchLock {

    private static final String redisHost = "10.0.5.86";

    private static final int port = 6381;

    private static JedisPoolConfig config;

    private static JedisPool pool;

    private static ExecutorService service;

    private static int ThLeng=10;

    private static CountDownLatch latch;

    private static AtomicInteger Countor = new AtomicInteger(0);
    static{
        //利用Redis连接池,保证多个线程利用多个连接,充分模拟并发性
        config = new JedisPoolConfig();
        config.setMaxIdle(10);
        config.setMaxWaitMillis(1000);
        config.setMaxTotal(30);
        pool = new JedisPool(config, redisHost, port);
        //利用ExecutorService 管理线程
        service = Executors.newFixedThreadPool(10);
        //CountDownLatch保证主线程在全部线程结束之后退出
        latch = new CountDownLatch(ThLeng);
    }

    public static void main(String args[]){
        int ThLeng = 10;
        String ThreadNamePrefix = "thread-";
        Jedis cli = pool.getResource();
        cli.del("redis_inc_key");//先删除既定的key
        cli.set("redis_inc_key", String.valueOf(1));//设定默认值
        for(int i =0;i<ThLeng;i++){
            Thread th = new Thread(new TestThread(pool));
            th.setName(ThreadNamePrefix+i);
            System.out.println(th.getName()+"inited...");
            service.submit(th);
        }
        service.shutdown();
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("all sub thread sucess");
        System.out.println("countor is "+Countor.get());
        String countStr = cli.get("redis_inc_key");
        System.out.println(countStr);
    }

    public static class TestThread implements Runnable {
        private String incKeyStr = "redis_inc_key";
        private Jedis cli;
        private JedisPool pool;
        public TestThread(JedisPool pool) {
            cli = pool.getResource();
            this.pool = pool;

        }
        public void run() {
            try{

                for (int i = 0; i < 100; i++) {
                    actomicAdd();
                }
            }catch(Exception e){
                pool.returnBrokenResource(cli);
            }
            finally{
                pool.returnResource(cli);
                latch.countDown();
            }
        }

        public void actomicAdd(){
            cli.watch(incKeyStr);
            boolean flag =true;
            while(flag){
                String countStr = cli.get("redis_inc_key");
                int countInt = Integer.parseInt(countStr);
                int expect = countInt+1;
                Transaction tx = cli.multi();                   
                tx.set(incKeyStr, String.valueOf(expect));
                List<Object> list = tx.exec();
                //如果事务失败了exec会返回null
                if(list==null){
                    System.out.println("multi shut down");
                    continue;
                }
                else{
                    //如果达到期望值那么结束while循环
                    flag=false;
                }
                System.out.println("my expect num is "+expect);         
                System.out.println("seting....");   
            }
            Countor.incrementAndGet();  
        }

    }

}

延伸难点

ok,那么我们以为那时候已经竣事了啊?其实并从未。。。为啥吧?
因为会现出如下那样一种相比较极端的情形:

不怕义务完毕之后去doing队列中去除message战败,然后去pending中除去也失利,因为有大概在职分扫描的时候,吧职责刚放入pending队列中,没等doing完成吗,pending中再一次放入的天职就被消费了。那么此时依旧是音信出现重复

那种情况下的极品消除方案是怎么着吧?正是消费端做好幂等性处理(其实像Ali的罗克etMq)也会出现消息再一次的情景(固然相当的低可能率),不过在Mq中,仿佛设计2个标准只发2遍的模型,是一件比较难的事务。

不过在multi的历程中只要locktest的值发生变化又会怎么?

发端达成

落实ack的话,(方今先不考虑集群版,只是单机版本)

  1. 自个儿得以用lpush做生产者,每一趟有新闻供给生产的时候,就发送3个message到pending队列中。
  2. brpoplpush做消费者,每回取到音讯的时候实行作业消费。在费用的同时吧音讯放到另五个doing的类别中
  3. 老是消费者完毕职责,从doing队列中删去职分msg,用来报告那么些音信被成功消费掉了
  4. 下一场开3个线程去定时轮询查doing中,借使一定时间(架设我们的message完成了大家的协商,message中包括职分起先的年华戳),那几个职分还没被消费成功,那么就把这一个doing队列的不胜就再也塞到pending的队列里
127.0.0.1:6379> set locktest 1
OK
127.0.0.1:6379> get locktest
"1"
127.0.0.1:6379> watch locktest
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set locktest 3
QUEUED
127.0.0.1:6379> exec
(nil)
127.0.0.1:6379> get locktest
"2"
127.0.0.1:6379> 

意识难题

而是此时也许会出现那样的题材,作者轮询doing的队列在取任务的时候大概因为小编消费者的职务因为某个原因做的慢了些,那么此时就会被另行塞会pending队列里,可是过两秒笔者的doing确实消费完了。

那么怎么消除那么些题材呢?

竭泽而渔措施实际上很粗大略,正是上边的进行步骤3的时候,要是从doing队列实行删除的时候,尽管重临值表示删除退步以来,那么表明我们的职务被系统认为过期了,他被赛入pending中了,那么大家只须求在这几个时候去pending中另行删除那一个message音讯即可

对此时常开发Web的Coder们,日常会有那样的供给,就是在多机的分布式环境下,有时候须求限制多台机器上的恳求修改同一份能源。对于单机的条件下,我们熟视无睹能够用联合如故锁去幸免四线程下的竞态条件。以java为例,我们得以用synchronized大概ReentrantLock,去做财富访问的一块。但那是JVM和操作系统提须要大家的特性,不过对于分布式环境下我们并未那么些方便人民群众条件。所以大家须求引入一个外表的Observer去贯彻那样的一个分布式锁,Zookeeper是3个相比较好的缓解方案,可是Zookeeper依然比较重的,大家可以用Redis完成如此一个锁。
乐天锁基于CAS思想,是不有所互斥性,不会发生锁等待而消功耗源,不过必要频仍的重试,但也是因为重试的建制,能比较快的响应。在落到实处CAS此前,必要了然一下Redis的思想政治工作机制。
Redis事务:
咱俩得以用Mysql事务机制来通晓Redis的事情机制,但也截然不一样,Mysql的事情的格局如下:
openSession()
update()
insert()
commit()
如果在update和insert之间出现谬误,那么会触发rollback(),Redis的事务用到了MULTI和EXEC命令,事务的花样如下:
MULTI
SET
HSET
EXEC
和Mysql的事务分裂,Redis会将全数EXEC命令从前的通令放入二个QUEUE中,当遇到EXEC时批量履行QUEUE中的命令,可是Redis的事体是不援助回滚的,它只是逐一的执行命令,并批量回来结果,可是对于极端气象下,事务在并未完全执行完时宕机,导致事情日志只写入部分,那样在重启时会发生错误,用aof的修复工具修复后得以展开运维。
在打听了事情机制后,我们还不足以达成乐观锁,还亟需精通一个下令——沃特ch,Watch命令能够监察和控制Redis中的七个key,当Key发生变化时停下事务的交付。先看二个不错的例证:

传统MQ的缺点

MQ基本上和缓存一样是住家必备之良药。可是音信队列固然首要,可是还要其实是蛮重的三个组件。例如咱们在用rabbitMq的话,大家需求为它搭建1个服务端,借使设想到可用性,那么我们必要为服务端建立三个集群,同时,大家只要线上难点或然还亟需在Mq中做查找,那么这么些干活儿就恐怕加大大家完全的工作量。

127.0.0.1:6379> set locktest 1
OK
127.0.0.1:6379> get locktest
"1"
127.0.0.1:6379> watch locktest
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set locktest 3
QUEUED
127.0.0.1:6379> exec
1) OK
127.0.0.1:6379> get locktest
"3"
127.0.0.1:6379> 

利用redis来实现MQ

为此就想能还是不能够先简单的经过Redis来促成音讯队列呢?不考虑PubSub、分布式、持久化、事务等繁杂的情事。就像JDK的种种Queue一样。答案当然是足以的,因为Redis提供的list数据结构就极度适合做新闻队列。大家兴许会发现,网上有好多redis的音信队列,不过近日停止,小编从未察觉一个音信队列是富有ack机制的。

这里大家会讲述怎么使用list的api中的lpush/brpoplpush来完结1个有所ack机制的新闻队列

那里大家用另三个Client在Multi然后将locktest修改为2,课件在举行工作的时候回来为nil,表示执行破产。
那么我们就足以用上述三种命令达成三个有望锁,代码如下:

音信队列(MQ)

信任大家对MQ本条词都不会面生,不管用过大概没用过的,大多会对他有早晚的询问,
那正是说信息队列有啥便宜吗

  1. 解耦(接触服务时期的耦合度关系)
  2. 削峰(例如笔者有些打折活动在某些时间点有特别大的流量涌入,这么些时候用Mq做缓存是最好的章程了)
  3. 异步化(例如有个别服务是自个儿不要求在同步链中展开调用的,那么能够用mq来做1个异步消费)

深层延伸的方案

上边的音信再次其实依旧有优化的后路,具体的落实思路如下:

  1. 优化扫描的模型,吧扫描doing过期职分变成1个延迟扫描(如用delayedQueue完结延迟职务扫描)
  2. 澳门美高梅手机网站,啊各个执行的职分模型用ExecutorService来保管,存款和储蓄正在实行的Future
  3. 每回扫描到过期的天职就去内存中查找那一个职责的Future是或不是留存,假使存在则不须求吧doing的message放到pending中
  4. 只要急需超时机制以来,找到呼应的Future并且打消当前职分的举办,并把前面实施的操作进行业务回滚/rollback,把message放到pending中

而是本身并不推荐这一套方案,因为这一套方案过于复杂,自身正是否大家用redis作为新闻队列的初衷。

总结

redis作为音讯队列是有非常的大的局限性的,自身作为八个以缓存/内部存款和储蓄器存款和储蓄为主的东西,只是因为一些api上的性状,我们得以落实三个简约的队列服务,本身大家要采纳好工作的选料,灵活的应用redis的MQ帮忙,才能兑现1个好的服务。

基于上述思想的代码实践本身一度停放了github上,部分代码还在做成人中学。
github地址 :
https://github.com/wgd12389/redisses/

发表评论

电子邮件地址不会被公开。 必填项已用*标注