Cache是基于python的线程缓存插件,类似于redis的key,value形式的线程化数据存储功能,使用方便,相对于不是很复杂的缓存任务,Cache完全可以满足需求

使用方法:



Cache().setex(k, v, timeout)  #设置缓存及过期时间
Cache().get(k)  #获取缓存



项目地址: github.com

多线程接收数据并缓存到BlockingQueue，然后多线程消费缓存数据
这个例子如果稍微改造，可以实现按照数据类型区分缓存
请大家看看， 有没有什么问题， 请指出，谢谢 。
package com.ygd.learn.threadTest.lock1.dtc;

import java.text.MessageFormat;
import java.util.concurrent.*;

/**
 * Description
 * 模拟一个接口多线程接收数据，然后缓存到blockingQueue中， 然后多线程消费缓存在blockingQueue中的数据
 * 其中每个线程处理一个缓存的blockingQueue的数据， 也就是blockingQuere的数量和线程的接收/消费的线程数相同
 * Created by ygd on 2017/9/1.
 */
public class DtcTest {
    ConcurrentMap<Integer,ArrayBlockingQueue<String>> concurrentMap = new ConcurrentHashMap<Integer,ArrayBlockingQueue<String>>();// 全局的。用来存放缓存数据的blockingQueue
    private final int blockQueueCount = 20 ;// blockQueue的数量，设计的是一个线程一个blockQueue ,来存放接收的数据
    private final int blockQuerySize = 200 ;// 每个blockQueue的大小

    public static void main(String[] args) {

        DtcTest dtcTest = new DtcTest();
        ReceiveDataCache receiveDataCache = dtcTest.new ReceiveDataCache();
        receiveDataCache.execute();

        SaveData saveData = dtcTest.new SaveData();
        saveData.execute();
    }

    /**
     * 消费blockingQueue中的数据
     */
    class SaveData{


        public void execute(){
            ExecutorService executorService = Executors.newFixedThreadPool(blockQueueCount);
            for(int i = 0 ;i <blockQueueCount ; i++){
                ExecuteClass executeClass = new ExecuteClass();
                executeClass.setBlockQueueKey(i);
                executorService.submit(executeClass);
            }
        }

        class ExecuteClass implements Callable<String>{
            private int blockQueueKey ;
            @Override
            public String call() throws Exception {
                ArrayBlockingQueue<String>  arrayBlockingQueue = concurrentMap.get(this.getBlockQueueKey());
                for(;;){// 无限循环从blockQueue中取数据
                    try {
                        String data = arrayBlockingQueue.poll(1,TimeUnit.SECONDS);
                        if(null==data){
                            System.out.println(MessageFormat.format("currentThread {0} had no data ",Thread.currentThread().getName()));
                        }else{
//                            System.out.println(data+"-----");
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }

            public int getBlockQueueKey() {
                return blockQueueKey;
            }

            public void setBlockQueueKey(int blockQueueKey) {
                this.blockQueueKey = blockQueueKey;
            }
        }
    }


    /**
     * 接收数据的cache类
     */
    class ReceiveDataCache{

        public ReceiveDataCache(){
            this.init();
        }

        private void init(){
            /**
             * 1、创建一个concurrentMap来存放系统中的所有BlockingQueue，系统中一共有200个BlockQuere
             * 2、每个BlockQueue的长度是200
             * 以上两个数据结构就是接收数据的缓存结构
             */
            for(int i = 0 ;i <blockQueueCount ; i++){
                concurrentMap.put(i,new ArrayBlockingQueue<String>(blockQuerySize));
            }
        }

        public void execute(){
            ExecutorService executorService = Executors.newFixedThreadPool(blockQueueCount);
            for(int i = 0  ;i <blockQueueCount ; i++){
                ExecuteClass executeClass = new ExecuteClass();
                executeClass.setBlockQueueKey(i);
                executorService.execute(executeClass);
            }
        }


        class ExecuteClass implements Runnable{
            private int blockQueueKey ;
            @Override
            public void run() {
                ArrayBlockingQueue<String> arrayBlockingQueue = concurrentMap.get(this.getBlockQueueKey());
                int size = 0 ;
                for(;;){// 无限循环向blockQueue中放数据
                    try {
                        size ++ ;
                        String str = "block-"+this.getBlockQueueKey()+"thread-"+Thread.currentThread().getName()+"-size-"+size;
                        Thread.sleep(1000);
                        boolean result = arrayBlockingQueue.offer(str);
                        if(!result){
                            System.out.println(MessageFormat.format("currenThread {0} offer blockQueue {1} had full",Thread.currentThread().getName(),this.getBlockQueueKey()));

                        }else{
//                            System.out.println(str);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }

            public int getBlockQueueKey() {
                return blockQueueKey;
            }

            public void setBlockQueueKey(int blockQueueKey) {
                this.blockQueueKey = blockQueueKey;
            }
        }


    }
}

应用场景
程序开发中经常遇到一种情况：一个线程收数据(例如socket->recv)，一个线程处理数据。
这种情况下，我通常的处理方式是一个公共的buff、一个锁，接收线程收到数据后，加锁拷贝内存记录偏移。另一个线程判断偏移大于零时呢，加锁，内存拷贝出来，然后处理。
这时就设计，需要加锁，如果高速数据接收(如万兆网)，可能导致接收不及时丢包。
进而需要设计一种无锁的缓存
无锁缓存
思路


1.一个接收数据线程，一个处理数据线程

2.定义100个10M的缓存，每个缓存有一个枚举值状态，包括：空、在写入、在读取

3.这样的话，收数据线程，跟这个100个缓存的状态往里面写​，数据处理线程，根据100个缓存的状态处理


4.这样达到没有锁的效果​
具体的缓存个数，缓存大小可以根据实际需要定义。
缓存队列使用的话，应该顺序使用，到队列头部，然后从头开始，若缓存状态为在读入，那么说明处理数据太慢，数据处理不过来，应该有异常上报机制。


代码 

strcut{
enum bufState{empty,writeing,reading;};
buffState  st;
char * buff;
int offset;
} buff;

buff   buffQue[100];

createBuffQue();

线程通信有两种方式：共享内存与消息传递。

共享内存即多个线程共享程序的公共变量，通过变量状态的读写来进行隐式共享；消息传递则是线程之间没有公共变量，而是通过消息传递来进行显示的通信。而在Java中，采用的是共享内存来进行线程间通信(多线程共享变量方式 )
Java中线程间通信由Java内存模型（JMM）来控制，JMM决定了共享变量的写入何时对另一个线程可见。JMM的抽象示意图如下：

可以看到，每个线程都是有一个共享变量的副本，每次访问都是先访问共享变量副本，如果没有做出同步操作，那么可能会导致一些内存一致性问题。下面做一个实验。

实验性质：验证性实验

实验目的：验证内存一致性问题真实存在。

实验步骤：1. 创建一个读线程，对共享变量进行读取操作，并打印读取的结果；2.创建一个写线程，对共享变量进行修改，并打印修改后的结果；3.观察打印结果
共享变量：
public class Model {
	public int i;
}


读线程：
public class ReadThread extends Thread {
	Model model;

	public ReadThread(Model model) {
		this.model = model;
	}
	@Override
	public void run() {
		while (true) {
			System.out.println(this.getName() + ":"
					+ System.currentTimeMillis() + ":" + model.i);
		}
	}
}

写线程：
public class WriteThread extends Thread {
	Model model;

	public WriteThread(Model model) {
		this.model = model;
	}

	@Override
	public void run() {
		while (true) {
			model.i++;
			System.out.println(this.getName() + ":"
					+ System.currentTimeMillis() + ":" + model.i);
		}
	}
}

主函数：
public class Main {
	public static void main(String[] args) throws IOException {
		File file = new File("E://log.txt");
		if(!file.exists()){
			file.createNewFile();
		}
		System.setOut(new PrintStream(file));
		
		
		Model model = new Model();
		model.i = 1;

		ReadThread readThead = new ReadThread(model);
		readThead.setName("readThread");

		WriteThread writeThread = new WriteThread(model);
		writeThread.setName("writeThread");

		readThead.start();
		writeThread.start();
	}
}

测试结果（节选有问题片段）：

readThread:1444028570784:2

writeThread:1444028570784:2

readThread:1444028570784:2

writeThread:1444028570784:3

writeThread:1444028570784:4

readThread:1444028570784:3

writeThread:1444028570784:5

readThread:1444028570784:5

writeThread:1444028570784:6

readThread:1444028570784:6

注意第七行位置，writeThread已经读到i的值变为4，而readThread读取到的值仍为3.
一起学习交流呀