数据结构与算法-散列表

Posted on 2019-06-15 | Edited on 2022-09-21 | In 数据结构与算法 , 数据结构

散列表的原理

底层基于数组实现，数据通过hash函数计算出数组下标写入到数组中，查找通过hash函数找到数据。理论上删除、查找的效率O(1)。

散列表解决hash冲突

当key1不等于key2，但是hash(key1)==hash(key2)出现散列的冲突，一般有如下的方式解决。散列冲突可以用loadFactor表示：

1 2	locafactor = 元素个数/数组长度

链表

将散列的hash方法的结果对应成一个bucket或者slot，每个bucket是一个链表。

写入：当hash冲突后在链表后面追加一个元素。时间复杂度O(1)
读取：hash找到bucket，遍历链表，时间复杂度O(1)
删除：hash找到bucket，遍历链表，删除，时间复杂度O(1)

比较理想的情况下是元素个数=槽数，如果hash冲突比较大会退化成链表

优点：对内存要求低，只有写入数据才会在链表中加1；hash冲突代价小，增删改几本上是O(1)
缺点：不是连续空间无法很好利用cpu缓存；链表包含下一个数据的指针序列化成本高
场景：数据量大，负载因子高，频繁修改的场景。HashMap

开放寻址

写入：比如key1和key2冲突，key2写入时候发现这个数组下标对应的元素有key1了这时候继续想后遍历知道有一个为空的字段，写入
读取：找到下标元素后比对如果不相等向后遍历直到找到相等的元素或者第一个为空的元素为止
删除：由于上面说的查找需要遍历到为空的元素，所以需要用到标记删除法。

优点：内存是连续地址很好的用到cpu的缓存；不包含指针序列化成本低
缺点：内存要求高；hash冲突成本高，查找、删除、修改可能要遍历数组
场景：数据量小，负载因子不高的场景。ThreadLocalMap

构建一个hash散列应该注意的问题

hash算法高效

自身算法效率要高，否则资源会大量浪费在hashkey的计算上。
算法要均匀尽量避免hash冲突，防止退化成链表

具体算法有：直接寻址法、平方取中法、折叠法、随机数法

如何避免装载因子多大

由于装载因子都大会导致hash冲突几率变高，链表过长，这时候如果是动态的散列表我们要进行”动态扩容“。

一般会设定一个负载因子的阈值打到阈值后开始扩容，扩容到原来的两倍，负载因子值会下降一半。动态扩容的事件复杂度是O(n)
如果系统对于内存的使用不敏感，对于响应敏感，为了防止hash冲突可以把负载因子值调低
如果系统对于内存的使用敏感，对于响应速度不敏感，可以将这个负载因子阈值调高，甚至可以大于1。

如何避免低效扩容

在散列表很大时候扩容有可能阻塞正常的读写操作，比如：1G的数据扩容到2G。这时候我们可以采用如下方案：

在额外申请一块2G的空间，新的写入操作往新的散列表写，同时将一个老的数据写过来，几次之后老的数据就会完成迁移
读的时候先从新的里面读取如果没有值在从老的数据里面读取

hash表和其他数据结构共用的案例

实现高效的lru算法

如果实现LRU算法用双向链表，访问、更新一个元素需要遍历链表，时间复杂度是O(n),如果想提高效率需要用到散列表配合

散列表+双向链表


type LRUNode struct {
  Prev *LRUNode //Lru双向链表
  Next *LRUNode //Lru双向链表
  Data interface{}
  Hnext *LRUNode  //散列值
}

访问：其中prev+next是双向链表用来实现LRU功能，每次访问节点先从hashTable中获取数据，然后将数据移到链表尾部。时间复杂度O(1)
写入：插入到散列中Hnext是用来维护散列的拉链，这里用的是链表来解决hash冲突，同时插入链表的尾部。时间复杂度O(1)
删除：找到数据从散列中删除，然后在从链表中删除。时间复杂度O(1)

java的LinkedHashMap也是用这个思想实现的。

redis的sortset

用跳表来实现，用score查找、排序、按照sorce范围查找。
用散列来实现，用key查找、删除、插入。

数据结构与算法-跳表

Posted on 2019-06-15 | Edited on 2022-09-21 | In 数据结构与算法 , 数据结构

跳表

数组的特点是支持随机访问，随机访问方便，但是插入、删除、修改比较麻烦【时间复杂度是O(N)】
链表的特点是不支持随机访问【时间复杂度是O(N)】，但是插入、删除、修改比较方便【时间复杂度是O(1)】

如果业务在效率上要兼顾随机访问和修改，是否有好的方案呢，就是我们今天的主角–跳表（skiplist），甚至在某些方便可以取代红黑叔

跳表的原理以及事件复杂度

原理

跳表的底层依赖是链表，但是在链表上层加了索引层。每个节点都包含一个down指针，索引层的节点的down指针指向下一层的节点,如图:
avator
查找时候先从上遍历索引层确定范围，然后通过down指针逐层在缩小的数据范围内找到数据。

复杂度

时间复杂度
查询：O(mlogN),m是每层遍历都少个节点
修改：O(logN)
空间复杂度
O(n):由于链表增加了索引层，索引层一般是一个等比数列，比如说：没2个节点建立一个索引，所以空间复杂度是n/2+n/4+n/8+n+4+2=n-2。

跳表最坏情况

动态更新如果增修改数据层，或者添加不平很会导致跳表退化成链表如图，所以需要一个随机函数，随机的将节点添加到任意一层索引层，比如随机函数生成了k，要讲节点添加到1~k层
avator

使用场景

数据有序、能利用到2分发对数据进行查找、同时兼顾修改的效率（redsi的sortSet）。

JAVA并发案例分析-高性能队列Disruptor的设计

Posted on 2019-06-08 | Edited on 2022-09-21 | In java , 并发

Disruptor的使用方法

生产消费针对对象event,定义Event
构造Disruptor时候需要实现一个EventFactory，这里是LongEvent::new
消费者要注册是一个handleEvent
生产者要通过publishEvent

// 自定义 Event
class LongEvent {
  private long value;
  public void set(long value) {
    this.value = value;
  }
}
// 指定 RingBuffer 大小,
// 必须是 2 的 N 次方
int bufferSize = 1024;

// 构建 Disruptor
Disruptor<LongEvent> disruptor
  = new Disruptor<>(
    LongEvent::new,
    bufferSize,
    DaemonThreadFactory.INSTANCE);

// 注册事件处理器
disruptor.handleEventsWith(
  (event, sequence, endOfBatch) ->
    System.out.println("E: "+event));

// 启动 Disruptor
disruptor.start();

// 获取 RingBuffer
RingBuffer<LongEvent> ringBuffer
  = disruptor.getRingBuffer();
// 生产 Event
ByteBuffer bb = ByteBuffer.allocate(8);
for (long l = 0; true; l++){
  bb.putLong(0, l);
  // 生产者生产消息
  ringBuffer.publishEvent(
    (event, sequence, buffer) -> event.set(buffer.getLong(0)), bb);
  Thread.sleep(1000);
}

Disruptor高效的要点

数据结构采用RingBuffer并且针对RingBuffer做了如下的优化，初始化时候利用EventFactory的newInstance方法创建所有的元素，
1. 由于一起创建所以这些元素在内存地址上是连续的，在消费元素时候有效的理由了CPU缓存（程序的局部性）当消费元素a的时候，a+1会加载到CPU的cashe中。
2. 在生产元素时候，利用setEvent这种方式重用对象，避免重新创建对象频繁的gc
解决伪缓存的方式：利用内存填充的方式防止变量共享一个缓存行，在无锁的并发情况下导致缓存行重复失效。
CAS的无锁设计高效生产消费队列

// 生产者获取 n 个写入位置
do {
  //cursor 类似于入队索引，指的是上次生产到这里
  current = cursor.get();
  // 目标是在生产 n 个
  next = current + n;
  // 减掉一个循环
  long wrapPoint = next - bufferSize;
  // 获取上一次的最小消费位置
  long cachedGatingSequence = gatingSequenceCache.get();
  // 没有足够的空余位置
  if (wrapPoint>cachedGatingSequence || cachedGatingSequence>current){
    // 重新计算所有消费者里面的最小值位置
    long gatingSequence = Util.getMinimumSequence(
        gatingSequences, current);
    // 仍然没有足够的空余位置，出让 CPU 使用权，重新执行下一循环
    if (wrapPoint > gatingSequence){
      LockSupport.parkNanos(1);
      continue;
    }
    // 从新设置上一次的最小消费位置
    gatingSequenceCache.set(gatingSequence);
  } else if (cursor.compareAndSet(current, next)){
    // 获取写入位置成功，跳出循环
    break;
  }
} while (true);

JAVA并发案例分析-Netty于Reactor模式

Posted on 2019-06-06 | Edited on 2022-09-21 | In java , 并发

BIO的问题

痛点：因为JAVA和操作系统的BIO操作是阻塞的，也就是说在读操作的时候线程会被阻塞住。
解决方案：

利用Thread-Per-Message模式每个IO操作都用一个线程。
问题：在java中由于线程对象较重，在链接不多的时候还可以一旦链接很多的场景很块就会出现问题。

Nio和WorkThread模式

我们可以利用WorkThread模式采用线程池来处理这种情况么？答案是BIO不可以因为BIO的IO操作会阻塞线程。这时候就要用到NIO。
NIO的IO操作是不会阻塞线程的：
以select模型为例子，简单来说：当一个IO请求发出后，不会阻塞当前线程，用户态的selecter会一直轮询FD检查是否IO请求的数据是否准备好，内核态会去准备数据，一旦内核态的数据准备好，会通知FD已经准备好
这时候我们的WorkThread可以对这些准备好的FD去处理。

上述这种模式我们叫它Reactor模式

Reactor模式的示例

avator

Reactor：模式中最重要的类：
- 负责调用操作系统的Selecter.select()函数，获取准备好的io-handlers
- 遍历io-handlers交给EventHandler
- 注册、移除EventHandler
EventHandler:io请求的逻辑处理类
Selecter：封装操作系统IO模型的选择器epoll、select
io-heandler：封装操作系统io请求类

netty

avator

如图：netty中的Reactor类就是EventLoop，Soekcet和Eventloop是稳定的多对1关系，而线程和EventLoop是稳定的1对1关系。绑定关系见源码的register
而eventloop是封装在一个EventLoopGroup中的并且在其中实现了负载均衡。
注意：netty的服务端往往需要俩个EventLoopGroup

boosGroup用于处理io的链接请求，如果监听一个端口值需要分配一个eventloop否则会浪费
workGroup处理具体io的操作。

原因是，操作系统的io决定的创建链接后会生成一个新的socker，为了性能我们会将这个socket放在其他的group中。

JAVA并发案例分析-Guava中的限流与高效令牌桶算法

Posted on 2019-06-06 | Edited on 2022-09-21 | In java , 并发

令牌桶算法的作用

主要使用场景–限流。服务器在处理上游的请求时候为了防止资深被打垮往往需要限流这种保护机制，其中有一种算法是令牌桶算法。
即：在指定的时间内允许一定的请求通过。

实现方案

方案1：生产者模式，消费者模式。用一个阻塞队列初始好令牌，请求来了从队列里消费，然后另外的线程按照一定速度往队列里增加令牌。缺点：在高并发场景下，线程调度的误差会变大，同时本身会创建线程调度会影响到线上的服务
方案2：因为令牌是按照事件产生的，我们可以利用时间关系来创建”令牌“
- 我们在桶中初始化令牌数，并且设置下一个产生令牌的时间next
- 如果当前请求时间now大于next，则计算产生的令牌，同时更新next的值为now。
- 取令牌，如果是令牌桶中取的不更新next，如果令牌桶没有了更新next时间
- 计算现在时间和next的差值，进行限流策略（阻塞或者返回错误）

代码示例


public class SimpleLimit {

    private long next = System.currentTimeMillis();
    private long intrval = 1_000_000_000;
    private long permitSize = 0;

    public SimpleLimit(long permitSize) {
        this.permitSize = permitSize;
    }

    //计算产生了多少令牌
    private void rsync(long now) {
        if (now > next) {
            long permits = (now - next) / intrval;

            permitSize = Math.min(permitSize, permitSize + permits);

            next = now;
        }
    }


    private synchronized long reserve(long now) {
        rsync(now);

        long at = next;

        //如果从桶里取的next不更新，如果桶里没有了next更新
        long pf = Math.min(1, permitSize);
        long nr = 1 - pf;
        next = next + nr * intrval;

        permitSize -= pf;
        return at;
    }

    public void acquire() {
        long now = System.nanoTime();

        long at = reserve(now);

        //判断是否应该限流，因为桶没了next更新所以at>now这时候休眠
        long waitTime = Math.max(at - now, 0);
        if (waitTime > 0) {
            try {
                TimeUnit.NANOSECONDS.sleep(waitTime);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

    }
}

生产者消费者模式--解耦整个业务

Posted on 2019-06-02 | Edited on 2022-09-21 | In java , 并发

生产者消费者模式的使用场景和作用

业务解耦：产者负责生产任务丢给队列，消费者负责从队列中获取任务做逻辑
削峰：串行的业务往往会导致服务夯住，如果采用生产者消费者，消费者负责将业务丢给队列即可返回，消费者可独立扩容多个来负责消费任务

消费者模式的几种实现

批量消费

我们可以从队列中一个个的消费任务，但是在某些场景下我们其实批量的执行任务会更提高效率，比如异步写入数据的逻辑，一条条会建很多链接，我们可以一次拉取N个任务一次执行，代码如下

public class BatchQueue<T> {

    private LinkedBlockingQueue<T> queue = new LinkedBlockingQueue<>();

    private int batchSize;

    private BatchJob<T> job;

    private ExecutorService pool;

    public BatchQueue(BatchJob<T> job, int batchSize, int consumerSize) {
        this.batchSize = batchSize;
        this.job = job;
        pool = Executors.newFixedThreadPool(consumerSize);
        pool.execute(() -> {
            while (true) {
                //异步消费
                List<T> tasks = pollTask();
                job.run(tasks);
            }
        });
    }

    public void addTask(T task) {
        queue.add(task);
    }

    private List<T> pollTask() {
        List<T> list = new ArrayList<>();
        T task = null;
        try {
            task = queue.take();//减少循环第一次先等待
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        while (task != null && list.size() < batchSize) {

            //循环里用poll，满足batchSize或者为空返回
            list.add(task);
            task = queue.poll();
        }

        return list;
    }

    abstract static class BatchJob<T> {
        public abstract void run(List<T> task);
    }

    public static void main(String[] args) {
        BatchQueue<Object> queue = new BatchQueue<>(new BatchJob<Object>() {
            @Override
            public void run(List<Object> task) {
                System.out.println(task.size());
                System.out.println(task);
            }
        }, 3, 10);


        for (int i = 0; i < 100; i++) {
            queue.addTask("i=" + i);
        }
    }
}

俩阶段提交

之前讨论Mysql的时候，我们提到过Mysql脏页的俩阶段提交，我们也可以使用消费者模式来实现

private LinkedBlockingQueue<T> queue = new LinkedBlockingQueue<>();

private int index = 0;
private long ptf = System.currentTimeMillis();

public void addTask(T task) {
    queue.add(task);
    
    ExecutorService executorService=Executors.newFixedThreadPool(10);
    executorService.execute(()->{
        try {
            consumer();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
}

private void consumer() throws InterruptedException {
    Object obj = queue.poll(5, TimeUnit.SECONDS);

    while (true) {
        if (obj != null) {
            System.out.println("write");
            //write to
        }

        if (index <= 0) {
            return;
        }

        //级别等于Error or 500条 or >=5秒
        if (index == 500 || System.currentTimeMillis() - ptf >= 5000L) {
            System.out.println("flush");
            index = 0;
            ptf = 0L;
        }
    }
}

优雅的终止线程--俩阶段终止

Posted on 2019-06-02 | Edited on 2022-09-21 | In java , 并发

如何正确的终止线程

线程终止的方法往往是运行完成、或者异常终止。但是如何正确的让线程T1优雅终止线程T2呢？
我们的JAVA提供线程的interrupt方法用来通知如何停掉线程，但是在实现上要注意以下问题

interrupt会导致线程的wait和sleep抛出InterruptedException，在处理完该异常时候interrupt会被重置为false
第三方包有可能未正确处理interrupt，所以最佳实践是我们要自己处理线程停止标记位也就是下面的stop

也不能只依赖标记位，因为很可能这时候线程处于wait或者block状态，所以应该标记位+interrupt方法一起

  public class StopThread {

    private Thread work;

    private boolean stop = false;

    private AtomicBoolean lock = new AtomicBoolean(false);

    public void startJob() {
        if (lock.get()) {
            return;
        }
        lock.compareAndSet(false, true);
        work = new Thread(() -> {
            while (!stop) {
                System.out.println("======job run!!!====");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    //重置中断标记位
                    Thread.currentThread().interrupt();
                }
            }

            //释放锁
            lock.compareAndSet(true, false);
            System.out.println("======job stop====");
        });
        work.start();
    }


    public void stopJob() {
        //方式单纯的interrupt没有被正确处理，所以标志位置为终止
        stop = true;
        //防止线程睡眠
        work.interrupt();
    }

}

多线程分工模式--Thread-Per-Message模式

Posted on 2019-05-26 | Edited on 2022-09-21 | In java , 并发

从名字可以看出，为每一个任务或者消息分配一个线程，适用于类似httpServer的场景，为每一个请求分配一个线程去执行。但是在java中创建线程的成本比较高（线程创建耗时且线程占用的内存较大）有俩种取代方案

用线程池取代，利用线程的复用思想
用更轻量的线程方案取代，golang中的协程，java中loom中的Fiber

注：多线程分工模式下这个应该是最简单的方案了

java中的另一个模式WorkThread模式

java创建线程、以及线程切换的开销很重，所以我们可以采用workThread模式即用线程池，来复用线程。

workthrea的“坑”

由于线程池中的任务有相互依赖关系，导致线程池因为没有资源被死锁，见下面：

开辟线程池
避免线程池的共享

//L1、L2 阶段共用的线程池
ExecutorService es = Executors.
  newFixedThreadPool(2);
//L1 阶段的闭锁    
CountDownLatch l1=new CountDownLatch(2);
for (int i=0; i<2; i++){
  System.out.println("L1");
  // 执行 L1 阶段任务
  es.execute(()->{
    //L2 阶段的闭锁 
    CountDownLatch l2=new CountDownLatch(2);
    // 执行 L2 阶段子任务
    for (int j=0; j<2; j++){
      es.execute(()->{
        System.out.println("L2");
        l2.countDown();
      });
    }
    // 等待 L2 阶段任务执行完
    // for循环俩个线程都会在l2.await()时候阻塞住,因为线程池只有2个l1 l2共享线程池且有关联，所以没有资源了
    l2.await();
    l1.countDown();
  });
}
// 等着 L1 阶段任务执行完
l1.await();
System.out.println("end");

Balking模式：再谈线程安全的单例模式

Posted on 2019-05-26 | Edited on 2022-09-21 | In java , 并发

我们在双重检查模式的单例中最外层的if就是Balking模式的一种体现，即快速返回错误。这样不用每次都进入到synchronized加锁

Guarded Suspension模式：等待唤醒机制的规范实现

Posted on 2019-05-26 | Edited on 2022-09-21 | In java , 并发

前面提到了在RPC调用中的同步转异步的过程，见Dubbo的DefaultFeture。
我们把这种模式成为Guarded Suspension（守护挂起），又称多线程的if，有一下几个特点

创建一个GuardedObject包含ID和Response对象，并且在get是否判断Response对象是否为空，如果为空该线程wait住
在其他的线程对GuardedObject的Response赋值，并且notify。
主要get和setResponse的要加锁。

具体实现见下面

通过GuardObject.create创建一个实例
通过get()方法获取并且阻塞住
返回时候其他线程通过通过GuardObject.objArraived赋值

private static Map<String, GuardObject> container = new ConcurrentHashMap<>();
   private Lock lock = new ReentrantLock();
   private Condition wait = lock.newCondition();

   private String id;
   private T obj;

   private GuardObject(String id) {
       this.id = id;
   }

   /**
    * 创建并且放在容器中
    * @param id
    * @param <T>
    * @return
    */
   public static <T> GuardObject<T> create(String id) {
       GuardObject<T> guardObject = new GuardObject<>(id);
       container.put(id, guardObject);
       return guardObject;
   }

   //判断条件
   private boolean isArrived() {
       return obj != null;
   }

   //为response赋值
   public static <T> void objArraived(String id, T obj) {
       GuardObject<T> guardObject = container.get(id);
       if (guardObject == null) {
           throw new RuntimeException("not exists");
       }

       guardObject.arrived(obj);
   }

   //等待通知机制
   private void arrived(T obj){
       lock.lock();
       try {
           this.obj = obj;
           wait.signalAll();
       } finally {
           lock.unlock();
       }
   }


   
   public T get() {
       lock.lock();

       try {
           //最佳实现，没有response等待
           while (!isArrived()) {
               //等待，增加超时机制
               try {
                   wait.await(10L, TimeUnit.SECONDS);
               } catch (InterruptedException e) {
               }

               if (!isArrived()) {
                   throw new RuntimeException("timeout");
               }
           }
       } finally {
           lock.unlock();
       }
       return obj;
   }
   
   public String getId() {
       return id;
   }

   public T getObj() {
       return obj;
   }

Liu hao

励志当好厨子的程序员

RSS

GitHub E-Mail