消息中间件kafka即时发送消息对象,storm即时处理,数据入库频率五分钟一次;
2:开发思路
在每个jvm中使用单例对象,添加读写锁(java.util.concurrent.locks.ReentrantReadWriteLock)对象的使用,线程安全队列(java.util.concurrent.LinkedBlockingQueue<String>)对象,利用锁的特性来控制队列读写操作,达到读数据的时候 写操作堵塞。
下面贴上测试代码实现:
import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; /** * @author liqiankun * */ public class BlockQueue { private LinkedBlockingQueue<String> blockingQueue = new LinkedBlockingQueue<String>(); private ReadWriteLock rwl = new ReentrantReadWriteLock(); private static final BlockQueue QUEUE = new BlockQueue(); private BlockQueue() { } public static BlockQueue getQueue(){ return QUEUE; } public void putData(String s){ rwl.writeLock().lock();//取到写锁 try { blockingQueue.offer(s); System.out.println(Thread.currentThread().getName() + "写入数据:" + s); }catch (Exception e) { } finally { rwl.writeLock().unlock();// 释放写锁 } } public void execute(){ new Thread(){ @Override public void run() { while(true){ try { Thread.sleep(5000); } catch (InterruptedException e) { e.printStackTrace(); } rwl.readLock().lock(); try { //启动一个线程去处理数据,保证不影响下一次数据的读取操作 new Thread(){ private LinkedBlockingQueue<String> tempBlockingQueue = new LinkedBlockingQueue<String>(); { //将blockingQueue的数据引用赋值给tempBlockingQueue tempBlockingQueue = blockingQueue; //将blockingQueue重新new出一个对象。 blockingQueue = new LinkedBlockingQueue<String>(); } @Override public void run() { while(!tempBlockingQueue.isEmpty()){ String s = tempBlockingQueue.poll();// System.out.println(Thread.currentThread().getName()+"读取数据:"+s); } } }.start(); } catch (Exception e) { e.printStackTrace(); }finally{ rwl.readLock().unlock(); } } } }.start(); } }
测试类
import java.util.Random; import java.util.concurrent.LinkedBlockingQueue; /** * @author liqiankun * */ public class TestLock { private LinkedBlockingQueue<String> tempBlockingQueue = new LinkedBlockingQueue<String>(); private LinkedBlockingQueue<String> blockingQueue = new LinkedBlockingQueue<String>(); private static String[] str= {"1","2","3","4","5","6","7","8","9","10","11","12","13","14","15","16","17","18","19","20"}; public static void main(String[] args) { BlockQueue blockQueue = BlockQueue.getQueue(); blockQueue.execute(); //while(true){ for(int i=0;i<3;i++){ new Thread(){ @Override public void run() { for(int j=0;j<20;j++){ BlockQueue.getQueue().putData(Thread.currentThread().getName()+"-["+str[j]+"]"); try { Thread.sleep(new Random().nextInt(100)); } catch (InterruptedException e) { e.printStackTrace(); } } } }.start(); } try { Thread.sleep(new Random().nextInt(2000)); } catch (InterruptedException e) { e.printStackTrace(); } //} /*TestLock t = new TestLock(); t.blockingQueue.offer("1"); t.blockingQueue.offer("2"); t.blockingQueue.offer("3"); t.tempBlockingQueue = t.blockingQueue; t.blockingQueue = new LinkedBlockingQueue<String>(); System.out.println("--new 对象之后--------"); while(!t.blockingQueue.isEmpty()){ System.out.println(t.blockingQueue.poll()); } System.out.println("-----------"); while(!t.tempBlockingQueue.isEmpty()){ System.out.println("tempBlockingQueue"+t.tempBlockingQueue.poll()); } System.out.println("-----------"); */ } }
运行结果:
Thread-1写入数据:Thread-1-[1]
Thread-2写入数据:Thread-2-[1]
Thread-3写入数据:Thread-3-[1]
Thread-1写入数据:Thread-1-[2]
Thread-2写入数据:Thread-2-[2]
Thread-3写入数据:Thread-3-[2]
Thread-1写入数据:Thread-1-[3]
Thread-2写入数据:Thread-2-[3]
Thread-3写入数据:Thread-3-[3]
Thread-3写入数据:Thread-3-[4]
Thread-1写入数据:Thread-1-[4]
Thread-2写入数据:Thread-2-[4]
Thread-3写入数据:Thread-3-[5]
Thread-3写入数据:Thread-3-[6]
Thread-2写入数据:Thread-2-[5]
Thread-3写入数据:Thread-3-[7]
Thread-1写入数据:Thread-1-[5]
Thread-2写入数据:Thread-2-[6]
Thread-3写入数据:Thread-3-[8]
Thread-2写入数据:Thread-2-[7]
Thread-1写入数据:Thread-1-[6]
Thread-3写入数据:Thread-3-[9]
Thread-3写入数据:Thread-3-[10]
Thread-1写入数据:Thread-1-[7]
Thread-2写入数据:Thread-2-[8]
Thread-3写入数据:Thread-3-[11]
Thread-2写入数据:Thread-2-[9]
Thread-2写入数据:Thread-2-[10]
Thread-1写入数据:Thread-1-[8]
Thread-3写入数据:Thread-3-[12]
Thread-2写入数据:Thread-2-[11]
Thread-3写入数据:Thread-3-[13]
Thread-1写入数据:Thread-1-[9]
Thread-1写入数据:Thread-1-[10]
Thread-1写入数据:Thread-1-[11]
Thread-2写入数据:Thread-2-[12]
Thread-3写入数据:Thread-3-[14]
Thread-1写入数据:Thread-1-[12]
Thread-1写入数据:Thread-1-[13]
Thread-1写入数据:Thread-1-[14]
Thread-2写入数据:Thread-2-[13]
Thread-3写入数据:Thread-3-[15]
Thread-1写入数据:Thread-1-[15]
Thread-2写入数据:Thread-2-[14]
Thread-3写入数据:Thread-3-[16]
Thread-1写入数据:Thread-1-[16]
Thread-3写入数据:Thread-3-[17]
Thread-3写入数据:Thread-3-[18]
Thread-2写入数据:Thread-2-[15]
Thread-2写入数据:Thread-2-[16]
Thread-2写入数据:Thread-2-[17]
Thread-1写入数据:Thread-1-[17]
Thread-1写入数据:Thread-1-[18]
Thread-3写入数据:Thread-3-[19]
Thread-2写入数据:Thread-2-[18]
Thread-1写入数据:Thread-1-[19]
Thread-3写入数据:Thread-3-[20]
Thread-2写入数据:Thread-2-[19]
Thread-1写入数据:Thread-1-[20]
Thread-2写入数据:Thread-2-[20]
Thread-4读取数据:Thread-1-[1]
Thread-4读取数据:Thread-2-[1]
Thread-4读取数据:Thread-3-[1]
Thread-4读取数据:Thread-1-[2]
Thread-4读取数据:Thread-2-[2]
Thread-4读取数据:Thread-3-[2]
Thread-4读取数据:Thread-1-[3]
Thread-4读取数据:Thread-2-[3]
Thread-4读取数据:Thread-3-[3]
Thread-4读取数据:Thread-3-[4]
Thread-4读取数据:Thread-1-[4]
Thread-4读取数据:Thread-2-[4]
Thread-4读取数据:Thread-3-[5]
Thread-4读取数据:Thread-3-[6]
Thread-4读取数据:Thread-2-[5]
Thread-4读取数据:Thread-3-[7]
Thread-4读取数据:Thread-1-[5]
Thread-4读取数据:Thread-2-[6]
Thread-4读取数据:Thread-3-[8]
Thread-4读取数据:Thread-2-[7]
Thread-4读取数据:Thread-1-[6]
Thread-4读取数据:Thread-3-[9]
Thread-4读取数据:Thread-3-[10]
Thread-4读取数据:Thread-1-[7]
Thread-4读取数据:Thread-2-[8]
Thread-4读取数据:Thread-3-[11]
Thread-4读取数据:Thread-2-[9]
Thread-4读取数据:Thread-2-[10]
Thread-4读取数据:Thread-1-[8]
Thread-4读取数据:Thread-3-[12]
Thread-4读取数据:Thread-2-[11]
Thread-4读取数据:Thread-3-[13]
Thread-4读取数据:Thread-1-[9]
Thread-4读取数据:Thread-1-[10]
Thread-4读取数据:Thread-1-[11]
Thread-4读取数据:Thread-2-[12]
Thread-4读取数据:Thread-3-[14]
Thread-4读取数据:Thread-1-[12]
Thread-4读取数据:Thread-1-[13]
Thread-4读取数据:Thread-1-[14]
Thread-4读取数据:Thread-2-[13]
Thread-4读取数据:Thread-3-[15]
Thread-4读取数据:Thread-1-[15]
Thread-4读取数据:Thread-2-[14]
Thread-4读取数据:Thread-3-[16]
Thread-4读取数据:Thread-1-[16]
Thread-4读取数据:Thread-3-[17]
Thread-4读取数据:Thread-3-[18]
Thread-4读取数据:Thread-2-[15]
Thread-4读取数据:Thread-2-[16]
Thread-4读取数据:Thread-2-[17]
Thread-4读取数据:Thread-1-[17]
Thread-4读取数据:Thread-1-[18]
Thread-4读取数据:Thread-3-[19]
Thread-4读取数据:Thread-2-[18]
Thread-4读取数据:Thread-1-[19]
Thread-4读取数据:Thread-3-[20]
Thread-4读取数据:Thread-2-[19]
Thread-4读取数据:Thread-1-[20]
Thread-4读取数据:Thread-2-[20]
把写入数据和读取数据对比之后发现,写入的顺序和读取的顺序是一致的,所以测试通过