线程安全的无锁RingBuffer的实现

这里的线程安全，是指一个读线程和一个写线程，读写两个线程是安全的，而不是说多个读线程和多个写线程是安全的。。

 

 

在程序设计中，咱们有时会遇到这样的状况，一个线程将数据写到一个buffer中，另一个线程从中读数据。因此这里就有多线程竞争的问题。一般的解决办法是对竞争资源加锁。可是，通常加锁的损耗较高。其实，对于这样的一个线程写，一个线程读的特殊状况，能够以一种简单的无锁RingBuffer来实现。这样代码的运行效率很高。

本文借鉴了Disruptor项目代码。

代码我在github上放了一份，须要的同窗能够去下载（RingBuffer.java）。本文最后也会附上一份。

代码的基本原理以下。

如图所示，假定buffer的长度是bufferSize. 咱们设置两个指针。head指向的是下一次读的位置，而tail指向的是下一次写的位置。因为这里是环形buffer (ring buffer)，这里就有一个问题，怎样判断buffer是满或者空。这里采用的规则是，buffer的最后一个单元不存储数据。因此，若是head == tail，那么说明buffer为空。若是 head == tail + 1 (mod bufferSize)，那么说明buffer满了。

接下来就是最重要的内容了：怎样以无锁的方式进行线程安全的buffer的读写操做。基本原理是这样的。在进行读操做的时候，咱们只修改head的值，而在写操做的时候咱们只修改tail的值。在写操做时，咱们在写入内容到buffer以后才修改tail的值；而在进行读操做的时候，咱们会读取tail的值并将其赋值给copyTail。赋值操做是原子操做。因此在读到copyTail以后，从head到copyTail之间必定是有数据能够读的，不会出现数据没有写入就进行读操做的状况。一样的，读操做完成以后，才会修改head的数值；而在写操做以前会读取head的值判断是否有空间能够用来写数据。因此，这时候tail到head - 1之间必定是有空间能够写数据的，而不会出现一个位置的数据尚未读出就被写操做覆盖的状况。这样就保证了RingBuffer的线程安全性。

最后附上代码供参考。欢迎批评指正，也欢迎各类讨论！

 1 public class RingBuffer {
 2 
 3     private final static int bufferSize = 1024;
 4     private String[] buffer = new String[bufferSize];
 5     private int head = 0;
 6     private int tail = 0;
 7     
 8     private Boolean empty() {
 9         return head == tail;
10     }
11     private Boolean full() {
12         return (tail + 1) % bufferSize == head;
13     }
14     public Boolean put(String v) {
15         if (full()) {
16             return false;
17         }
18         buffer[tail] = v;
19         tail = (tail + 1) % bufferSize;
20         return true;
21     }
22     public String get() {
23         if (empty()) {
24             return null;
25         }
26         String result = buffer[head];
27         head = (head + 1) % bufferSize;
28         return result;
29     }
30     public String[] getAll() {
31         if (empty()) {
32             return new String[0];
33         }
34         int copyTail = tail;
35         int cnt = head < copyTail ? copyTail - head : bufferSize - head + copyTail;
36         String[] result = new String[cnt];
37         if (head < copyTail) {
38             for (int i = head; i < copyTail; i++) {
39                 result[i - head] = buffer[i];
40             }
41         } else {
42             for (int i = head; i < bufferSize; i++) {
43                 result[i - head] = buffer[i];
44             }
45             for (int i = 0; i < copyTail; i++) {
46                 result[bufferSize - head + i] = buffer[i];
47             }
48         }
49         head = copyTail;
50         return result;
51     }
52 }

 

 

 

标签:  ring buffer,  多线程,  无锁