第四节：Geo类型介绍以及Redis批量操作、事务、分布式锁

一. Geo类型

1. 类型说明

　　Geo 是 Redis 3.2 版本后新增的数据类型，用来保存兴趣点（POI，point of interest）的坐标信息。可以实现计算两 POI 之间的距离、获取一个点周边指定距离的 POI。

2. 常用Api

　　(1).GeoAdd：添加POI点

　　(2).GeoDistance：获取两点之间的最短距离

　　(3).GeoPosition：获取某个点的坐标

　　(4).GeoRadius：获取某个点(不一定是POI)周边xx米以外的点

　　(5).GeoRemove：删除某个点

代码分享：

           //1. 添加所有商店的地理位置
            db.GeoAdd("ShopsGeo", new GeoEntry(116.34039, 39.94218, "name1"));
            db.GeoAdd("ShopsGeo", new GeoEntry(116.340934, 39.942221, "name2"));
            db.GeoAdd("ShopsGeo", new GeoEntry(116.341082, 39.941025, "name3"));
            db.GeoAdd("ShopsGeo", new GeoEntry(116.340848, 39.937758, "name4"));
            db.GeoAdd("ShopsGeo", new GeoEntry(116.342982, 39.937325, "name5"));
            db.GeoAdd("ShopsGeo", new GeoEntry(116.340866, 39.936827, "name6"));

            //2. 计算商店name1和name2之间的距离(单位m)
            double? dist = db.GeoDistance("ShopsGeo", "name1", "name5", GeoUnit.Meters);

            //3. 获取name1商店的坐标
            GeoPosition? pos = db.GeoPosition("ShopsGeo", "name1");

            //4. 获取一个 name2 周边的200内的点： 
            GeoRadiusResult[] results = db.GeoRadius("ShopsGeo", "name2", 200, GeoUnit.Meters);
            foreach (GeoRadiusResult result in results)
            {
                Console.WriteLine("Id=" + result.Member + ",位置" + result.Position + "，距离" + result.Distance);
            }

            //5.  获取一个坐标(116.34092, 39.94223)（这个坐标不一定是 POI）周边的 POI： 
            GeoRadiusResult[] results2 = db.GeoRadius("ShopsGeo", 116.34092, 39.94223, 200, GeoUnit.Meters);
            foreach (GeoRadiusResult result in results2)
            {
                Console.WriteLine("Id=" + result.Member + ",位置" + result.Position + "，距离" + result.Distance);
            }

            //6. 删除
            bool d1 = db.GeoRemove("ShopsGeo", "name2");

3. 案例

　　地图上点相关的操作，方圆xx米内有多少个商店,某两个商店间的距离

 

二. Redis批量操作和事务

1. 批量操作

　　Batch会把所需要执行的命令打包成一条请求发到Redis，然后一起等待返回结果。这样批量操作的速度就大大提升。 利用CreateBatch和Execute方法，仅支持异步方法哦。

  public static void BatchDemo(IDatabase db)
  {
    IBatch batch = db.CreateBatch();
    batch.StringSetAsync("keen1", "111");
    batch.StringSetAsync("keen2", "222");
    batch.Execute();

  }

2. 事务

　　Redis的操作都是原子性单线程的，如果一次性操作很多，基本上每个方法都支持批量操作，但是如果操作的数据类型不同，可以使用Redis的事务进行包裹。 CreateTransaction和Execute方法

    public static void TransDemo(IDatabase db)
    {
            var trans = db.CreateTransaction();
            trans.StringSetAsync("keen1", "111");
            trans.StringSetAsync("keen2", 222);
            bool result = trans.Execute();
      }

3. 区别 

　　批量操作假设里面有一个出错，不会整体回滚，而事务要么都成功，要么都失败。

三. Redis分布式锁

1. 背景

　　在传统的单体项目中，即部署到单个IIS上，针对并发问题，比如进销存中的出库和入库问题，多个人同时操作，属于一个IIS进程中多个线程并发操作的问题，这个时候可以引入线程锁lock/Monitor等，轻松解决这类问题。但是随着业务量的逐渐增大，比如"秒杀业务"，肯定是集群，这个时候线程锁已经没用了，必须引入分布式锁。

　　常见的分布式锁有：数据库、zookeeper、redis。

2. 技术分析

　　秒杀业务集群同时访问DB，很容易出现超买超卖问题，如下图：　

 分析：

　

解决方案：

　在秒杀服务集群和DB之间引入Redis(或者Redis集群)，无论是Redis单体还是集群，分布式锁都是上一个解锁了下一个才继续加锁，引入Redis集群的目的是防止Redis崩溃，而不是加快速度，这样保证了最终到DB上上的操作是按顺序依次进行的，从而解决了超卖问题。如下图：

 

 3. 代码实战

　　StackExchange.Redis中加锁和解锁的api分别是：LockTake和LockRelease。 bool LockTake(RedisKey key, RedisValue value, TimeSpan expiry, CommandFlags flags = CommandFlags.None); 三个参数的含义分别是：

(1). 锁名

(2). 谁加的的锁

(3). 超时时间，过期自动释放，防止死锁。

代码如下：（加锁的时候要循环获取锁，直到获取为止）

        /// <summary>
        /// 分布式锁 业务
        /// </summary>
        /// <param name="db"></param>
        public static void DfsLockDemo(IDatabase db)
        {

            Lock(db);
            try
            {
                Console.WriteLine("业务执行中......");
                Thread.Sleep(8000);
                Console.WriteLine("业务执行完毕");
            }
            catch (Exception)
            {
                //释放锁
                UnLock(db);
            }
            finally
            {
                //释放锁
                UnLock(db);
            }
        }

        /// <summary>
        /// 加锁
        /// </summary>
        /// <param name="db"></param>
        public static void Lock(IDatabase db)
        {
            RedisValue token = Environment.MachineName;
            while (true)
            {
                bool flag = db.LockTake("myLock", token, TimeSpan.FromSeconds(10)); //10秒后自动释放
                if (flag)
                {
                    //表示获取成功，跳出while
                    break;
                }
                else
                {
                    Console.WriteLine("获取失败，继续获取");
                    Thread.Sleep(200);

                }
            }
        }
        /// <summary>
        /// 解锁
        /// </summary>
        /// <param name="db"></param>
        public static void UnLock(IDatabase db)
        {
            RedisValue token = Environment.MachineName;
            db.LockRelease("myLock", token);
        }

模拟两个项目运行效果：

秒杀案例其它思路或者详细解决方案见：   第六节：秒杀业务/超买超卖的几种解决思路

!

• 作       者 : Yaopengfei(姚鹏飞)
• 博客地址 : http://www.cnblogs.com/yaopengfei/
• 声     明1 : 本人才疏学浅，用郭德纲的话说“我是一个小学生”，如有错误，欢迎讨论，请勿谩骂^_^。
• 声     明2 : 原创博客请在转载时保留原文链接或在文章开头加上本人博客地址，否则保留追究法律责任的权利。