我的位置:首页 > Java开发>Java EE

Java队列和定时器Timer

时间:2019-07-24 09:25:00 来源:互联网 作者: 神秘的大神 字体:

   一: Queue详解

   Queue: 基本上,一个队列就是一个先入先出(FIFO)的数据结构

   Queue接口与List、Set同一级别,都是继承了Collection接口。LinkedList实现了Deque接 口。

  

  1)、没有实现的阻塞接口的LinkedList: 实现了java.util.Queue接口和java.util.AbstractQueue接口
  内置的不阻塞队列: PriorityQueue 和 ConcurrentLinkedQueue
  PriorityQueue 和 ConcurrentLinkedQueue 类在 Collection Framework 中加入两个具体集合实现。 
  PriorityQueue 类实质上维护了一个有序列表。加入到 Queue 中的元素根据它们的天然排序(通过其 java.util.Comparable 实现)或者根据传递给构造函数的 java.util.Comparator 实现来定位。
  ConcurrentLinkedQueue 是基于链接节点的、线程安全的队列。并发访问不需要同步。因为它在队列的尾部添加元素并从头部删除它们,所以只要不需要知道队列的大 小,ConcurrentLinkedQueue 对公共集合的共享访问就可以工作得很好。收集关于队列大小的信息会很慢,需要遍历队列。


  2)、实现阻塞接口的:
  java.util.concurrent 中加入了 BlockingQueue 接口和五个阻塞队列类。它实质上就是一种带有一点扭曲的 FIFO 数据结构。不是立即从队列中添加或者删除元素,线程执行操作阻塞,直到有空间或者元素可用。
五个队列所提供的各有不同:
  * ArrayBlockingQueue :一个由数组支持的有界队列。
  * LinkedBlockingQueue :一个由链接节点支持的可选有界队列。
  * PriorityBlockingQueue :一个由优先级堆支持的无界优先级队列。
  * DelayQueue :一个由优先级堆支持的、基于时间的调度队列。
  * SynchronousQueue :一个利用 BlockingQueue 接口的简单聚集(rendezvous)机制。

                    

  下表显示了jdk1.5中的阻塞队列的操作:

    add        增加一个元索                     如果队列已满,则抛出一个IIIegaISlabEepeplian异常
    remove   移除并返回队列头部的元素    如果队列为空,则抛出一个NoSuchElementException异常
    element  返回队列头部的元素             如果队列为空,则抛出一个NoSuchElementException异常
    offer       添加一个元素并返回true       如果队列已满,则返回false
    poll         移除并返问队列头部的元素    如果队列为空,则返回null
    peek       返回队列头部的元素             如果队列为空,则返回null
    put         添加一个元素                      如果队列满,则阻塞
    take        移除并返回队列头部的元素     如果队列为空,则阻塞

    3)、示例

 1 package com.svse.queue;
 3 import java.util.LinkedList;
 4 import java.util.Queue;
 7 import java.util.Timer;
 8 import java.util.TimerTask;
10 import com.svse.entity.Users;
11 
12 class TestQueue {
13 
14     static Queue<Users> queueUsers=new LinkedList<Users>();
15     static{
16         Users user1=new Users("201","张三","男","27","歌手");
17         Users user2=new Users("202","李思","女","26","演员");
18         queueUsers.add(user1);
19         queueUsers.add(user2);
20     }
21     
22     public void test1(){
23         
24         Queue<String> queue = new LinkedList<String>();
25         queue.offer("Hello");
26         queue.offer("World!");
27         queue.offer("你好!");
28         
29         System.out.println(queue.size());
30         
31         
32         while (queue.size() > 0) {
33             String element = queue.poll();
34             System.out.println(element);
35         }
36        System.out.println();
37        System.out.println(queue.size());
38     }
39     
40     //生产者
41     public void producerQueue(){
42         
43         System.out.println(queueUsers.size());
44     }
45     
46     //消费者
47     public void consumerQueue(){
48         Users u=null;
49          while((u=queueUsers.poll())!=null){
50                System.out.println(u+" ");
51             }
52        System.out.println();
53        System.out.println(queueUsers.size());
54     }
55    
69     
70     public static void main(String[] args) {
71         TestQueue tq=new TestQueue();
72         //tq.producerQueue();
73         //tq.consumerQueue();
74         
75         timerTest();
76         
77         
78         
79     }
80 
81 }

: 不怕难之BlockingQueue及其实现

  1)、 前言

       BlockingQueue即阻塞队列,它是基于ReentrantLock,依据它的基本原理,我们可以实现Web中的长连接聊天功能,当然其最常用的还是用于实现生产者与消费者模式,大致如下图所示:

                                  

 

 

        在Java中,BlockingQueue是一个接口,它的实现类有ArrayBlockingQueue、DelayQueue、 LinkedBlockingDeque、LinkedBlockingQueue、PriorityBlockingQueue、SynchronousQueue等,它们的区别主要体现在存储结构上或对元素操作上的不同,但是对于take与put操作的原理,却是类似的。

  2)、阻塞与非阻塞

  入队

offer(E e):如果队列没满,立即返回true; 如果队列满了,立即返回false-->不阻塞

put(E e):如果队列满了,一直阻塞,直到队列不满了或者线程被中断-->阻塞

offer(E e, long timeout, TimeUnit unit):在队尾插入一个元素,,如果队列已满,则进入等待,直到出现以下三种情况:-->阻塞

被唤醒

等待时间超时

当前线程被中断

  出队

poll():如果没有元素,直接返回null;如果有元素,出队

take():如果队列空了,一直阻塞,直到队列不为空或者线程被中断-->阻塞

poll(long timeout, TimeUnit unit):如果队列不空,出队;如果队列已空且已经超时,返回null;如果队列已空且时间未超时,则进入等待,直到出现以下三种情况:

被唤醒

等待时间超时

当前线程被中断

  3)、示例   

 1 package com.yao;
 2 import java.util.concurrent.ArrayBlockingQueue;
 3 import java.util.concurrent.BlockingQueue;
 4 import java.util.concurrent.ExecutorService;
 5 import java.util.concurrent.Executors;
 6 public class BlockingQueueTest {
 7  /**
 8  定义装苹果的篮子
 9   */
10  public static class Basket{
11   // 篮子,能够容纳3个苹果
12   BlockingQueue<String> basket = new ArrayBlockingQueue<String>(3);
13 
14   // 生产苹果,放入篮子
15   public void produce() throws InterruptedException{
16    // put方法放入一个苹果,若basket满了,等到basket有位置
17    basket.put("An apple");
18   }
19   // 消费苹果,从篮子中取走
20   public String consume() throws InterruptedException{
21    // get方法取出一个苹果,若basket为空,等到basket有苹果为止
22    String apple = basket.take();
23    return apple;
24   }
25 
26   public int getAppleNumber(){
27    return basket.size();
28   }
29 
30  }
31  // 测试方法
32  public static void testBasket() {
33   // 建立一个装苹果的篮子
34   final Basket basket = new Basket();
35   // 定义苹果生产者
36   class Producer implements Runnable {
37    public void run() {
38     try {
39      while (true) {
40       // 生产苹果
41       System.out.println("生产者准备生产苹果:" 
42         + System.currentTimeMillis());
43       basket.produce();
44       System.out.println("生产者生产苹果完毕:" 
45         + System.currentTimeMillis());
46       System.out.println("生产完后有苹果:"+basket.getAppleNumber()+"个");
47       // 休眠300ms
48       Thread.sleep(300);
49      }
50     } catch (InterruptedException ex) {
51     }
52    }
53   }
54   // 定义苹果消费者
55   class Consumer implements Runnable {
56    public void run() {
57     try {
58      while (true) {
59       // 消费苹果
60       System.out.println("消费者准备消费苹果:" 
61         + System.currentTimeMillis());
62       basket.consume();
63       System.out.println("消费者消费苹果完毕:" 
64         + System.currentTimeMillis());
65       System.out.println("消费完后有苹果:"+basket.getAppleNumber()+"个");
66       // 休眠1000ms
67       Thread.sleep(1000);
68      }
69     } catch (InterruptedException ex) {
70     }
71    }
72   }
73 
74   ExecutorService service = Executors.newCachedThreadPool();
75   Producer producer = new Producer();
76   Consumer consumer = new Consumer();
77   service.submit(producer);
78   service.submit(consumer);
79   // 程序运行10s后,所有任务停止
80   try {
81    Thread.sleep(10000);
82   } catch (InterruptedException e) {
83   }
84   service.shutdownNow();
85  }
86
87 public static void main(String[] args) { 88 BlockingQueueTest.testBasket(); 89 } 90 }

: 定时器之Timer

 1 package com.svse.queue;
 2 import java.util.Timer;
 3 import java.util.TimerTask;
 4 
 5 public class TestTimer {
 6 
 7 
 8     static int i=0;
 9     public static void timerTest(){
10         //创建一个定时器
11         Timer timer = new Timer();
12         //schedule方法是执行时间定时任务的方法
13         timer.schedule(new TimerTask() {
14             @Override
15             public void run() {
16                 i++;
17                 System.out.println("timerTest: "+i);
18             }
19         }, 1000, 60000); //第一个参数时间 从多少毫秒之后开始执行   第二个时间参数  间隔多少毫秒之后再执行 1分钟一次
20     }
21 
22     
23     public static void main(String[] args) {
24 
25         timerTest();
26         
27     }
28 
29 }