java生產者與消費者
❶ 由生產者/消費者問題看java多線程
生產者消費者問題是研究多線程程序時繞不開的問題 它的描述是有一塊生產者和消費者共享的有界緩沖區 生產者往緩沖區放入產品 消費者從緩沖區取走產品 這個過程可以無休止的執行 不能因緩沖區滿生產者放不進產品而終止 也不能因緩沖區空消費者無產品可取而終止
解決生產者消費者問題的方法有兩種 一種是採用某種機制保持生產者和消費者之間的同步 一種是在生產者和消費者之間建立一個管道 前一種有較高的效率並且可控制性較好 比較常用 後一種由於管道緩沖區不易控制及被傳輸數據對象不易封裝等原因 比較少用
同步問題的核心在於 CPU是按時間片輪詢的方式執行程序 我們無法知道某一個線程是否被執行 是否被搶占 是否結束等 因此生產者完全可能當緩沖區已滿的時候還在放入產品 消費者也完全可能當緩沖區為空時還在取出產品
現在同步問題的解決方法一般是採用信號或者加鎖機制 即生產者線程當緩沖區已滿時放棄自己的執行權 進入等待狀態 並通知消費者線程執行 消費者線程當緩沖區已空時放棄自己的執行權 進入等待狀態 並通知生產者線程執行 這樣一來就保持了線程的同步 並避免了線程間互相等待而進入死鎖狀態
JAVA語言提供了獨立於平台的線程機制 保持了 write once run anywhere 的特色 同時也提供了對同步機制的良好支持
在JAVA中 一共有四種方法支持同步 其中三個是同步方法 一個是管道方法
方法wait()/notify()
方法await()/signal()
阻塞隊列方法BlockingQueue
管道方法PipedInputStream/PipedOutputStream
下面我們看各個方法的實現
方法wait()/notify()
wait()和notify()是根類Object的兩個方法 也就意味著所有的JAVA類都會具有這個兩個方法 為什麼會被這樣設計呢？我們可以認為所有的對象默認都具有一個鎖 雖然我們看不到 也沒有辦法直接操作 但它是存在的
wait()方法表示 當緩沖區已滿或空時 生產者或消費者線程停止自己的執行 放棄鎖 使自己處於等待狀態 讓另一個線程開始執行
notify()方法表示 當生產者或消費者對緩沖區放入或取出一個產品時 向另一個線程發出可執行通知 同時放棄鎖 使自己處於等待狀態
下面是一個例子代碼
import java util LinkedList;
public class Sycn {
private LinkedList<Object> myList =new LinkedList<Object>();
private int MAX = ;
public Sycn (){
}
public void start(){
new Procer() start();
new Consumer() start();
}
public static void main(String[] args) throws Exception{
Sycn s = new Sycn ();
s start();
}
class Procer extends Thread{
public void run(){
while(true){
synchronized(myList){
try{
while(myList size() == MAX){
System out println( warning: it s full! );
myList wait();
}
Object o = new Object();
if(myList add(o)){
System out println( Procer: + o);
myList notify();
}
}catch(InterruptedException ie){
System out println( procer is interrupted! );
}
}
}
}
}
class Consumer extends Thread{
public void run(){
while(true){
synchronized(myList){
try{
while(myList size() == ){
System out println( warning: it s empty! );
myList wait();
}
Object o = myList removeLast();
System out println( Consumer: + o);
myList notify();
}catch(InterruptedException ie){
System out println( consumer is interrupted! );
}
}
}
}
}
}
方法await()/signal()
在JDK 以後 JAVA提供了新的更加健壯的線程處理機制 包括了同步 鎖定 線程池等等 它們可以實現更小粒度上的控制 await()和signal()就是其中用來做同步的兩種方法 它們的功能基本上和wait()/notify()相同 完全可以取代它們 但是它們和新引入的鎖定機制Lock直接掛鉤 具有更大的靈活性
下面是一個例子代碼
import java util LinkedList;
import ncurrent locks *;
public class Sycn {
private LinkedList<Object> myList = new LinkedList<Object>();
private int MAX = ;
private final Lock lock = new ReentrantLock();
private final Condition full = lock newCondition();
private final Condition empty = lock newCondition();
public Sycn (){
}
public void start(){
new Procer() start();
new Consumer() start();
}
public static void main(String[] args) throws Exception{
Sycn s = new Sycn ();
s start();
}
class Procer extends Thread{
public void run(){
while(true){
lock lock();
try{
while(myList size() == MAX){
System out println( warning: it s full! );
full await();
}
Object o = new Object();
if(myList add(o)){
System out println( Procer: + o);
empty signal();
}
}catch(InterruptedException ie){
System out println( procer is interrupted! );
}finally{
lock unlock();
}
}
}
}
class Consumer extends Thread{
public void run(){
while(true){
lock lock();
try{
while(myList size() == ){
System out println( warning: it s empty! );
empty await();
}
Object o = myList removeLast();
System out println( Consumer: + o);
full signal();
}catch(InterruptedException ie){
System out println( consumer is interrupted! );
}finally{
lock unlock();
}
}
}
}
}
阻塞隊列方法BlockingQueue
BlockingQueue也是JDK 的一部分 它是一個已經在內部實現了同步的隊列 實現方式採用的是我們的第 種await()/signal()方法 它可以在生成對象時指定容量大小
它用於阻塞操作的是put()和take()方法
put()方法類似於我們上面的生產者線程 容量最大時 自動阻塞
take()方法類似於我們上面的消費者線程 容量為 時 自動阻塞
下面是一個例子代碼
import ncurrent *;
public class Sycn {
private LinkedBlockingQueue<Object> queue = new LinkedBlockingQueue<Object>( );
private int MAX = ;
public Sycn (){
}
public void start(){
new Procer() start();
new Consumer() start();
}
public static void main(String[] args) throws Exception{
Sycn s = new Sycn ();
s start();
}
class Procer extends Thread{
public void run(){
while(true){
//synchronized(this){
try{
if(queue size() == MAX)
System out println( warning: it s full! );
Object o = new Object();
queue put(o);
System out println( Procer: + o);
}catch(InterruptedException e){
System out println( procer is interrupted! );
}
//}
}
}
}
class Consumer extends Thread{
public void run(){
while(true){
//synchronized(this){
try{
if(queue size() == )
System out println( warning: it s empty! );
Object o = queue take();
System out println( Consumer: + o);
}catch(InterruptedException e){
System out println( procer is interrupted! );
}
//}
}
}
}
}
你發現這個例子中的問題了嗎？
如果沒有 我建議你運行一下這段代碼 仔細觀察它的輸出 是不是有下面這個樣子的？為什麼會這樣呢？
…
warning: it s full!
Procer: java lang object@ e a
…
你可能會說這是因為put()和System out println()之間沒有同步造成的 我也這樣認為 我也這樣認為 但是你把run()中的synchronized前面的注釋去掉 重新編譯運行 有改觀嗎？沒有 為什麼？
這是因為 當緩沖區已滿 生產者在put()操作時 put()內部調用了await()方法 放棄了線程的執行 然後消費者線程執行 調用take()方法 take()內部調用了signal()方法 通知生產者線程可以執行 致使在消費者的println()還沒運行的情況下生產者的println()先被執行 所以有了上面的輸出 run()中的synchronized其實並沒有起什麼作用
對於BlockingQueue大家可以放心使用 這可不是它的問題 只是在它和別的對象之間的同步有問題
對於這種多重嵌套同步的問題 以後再談吧 歡迎大家討論啊！
管道方法PipedInputStream/PipedOutputStream
這個類位於java io包中 是解決同步問題的最簡單的辦法 一個線程將數據寫入管道 另一個線程從管道讀取數據 這樣便構成了一種生產者/消費者的緩沖區編程模式
下面是一個例子代碼 在這個代碼我沒有使用Object對象 而是簡單的讀寫位元組值 這是因為PipedInputStream/PipedOutputStream不允許傳輸對象 這是JAVA本身的一個bug 具體的大家可以看sun的解釋 _bug do?bug_id=
import java io *;
public class Sycn {
private PipedOutputStream pos;
private PipedInputStream pis;
//private ObjectOutputStream oos;
//private ObjectInputStream ois;
public Sycn (){
try{
pos = new PipedOutputStream();
pis = new PipedInputStream(pos);
//oos = new ObjectOutputStream(pos);
//ois = new ObjectInputStream(pis);
}catch(IOException e){
System out println(e);
}
}
public void start(){
new Procer() start();
new Consumer() start();
}
public static void main(String[] args) throws Exception{
Sycn s = new Sycn ();
s start();
}
class Procer extends Thread{
public void run() {
try{
while(true){
int b = (int) (Math random() * );
System out println( Procer: a byte the value is + b);
pos write(b);
pos flush();
//Object o = new MyObject();
//oos writeObject(o);
//oos flush();
//System out println( Procer: + o);
}
}catch(Exception e){
//System out println(e);
e printStackTrace();
}finally{
try{
pos close();
pis close();
//oos close();
//ois close();
}catch(IOException e){
System out println(e);
}
}
}
}
class Consumer extends Thread{
public void run(){
try{
while(true){
int b = pis read();
System out println( Consumer: a byte the value is + String valueOf(b));
//Object o = ois readObject();
//if(o != null)
//System out println( Consumer: + o);
}
}catch(Exception e){
//System out println(e);
e printStackTrace();
}finally{
try{
pos close();
pis close();
//oos close();
//ois close();
}catch(IOException e){
System out println(e);
}
}
}
}
//class MyObject implements Serializable {
//}
lishixin/Article/program/Java/gj/201311/27617
❷ Java 泛型中super T和extends T的區別
在Java泛型中，super T和extends T的區別在於它們各自對應的角色：生產者和消費者。PECS原則概括了這一點，即生產者（Procer）應使用extends，而消費者（Consumer）則使用super。
當作為生產者使用時，你需要一個列表能夠提供T類型的元素，也就是說，你打算從中讀取T類型的元素。在這種情況下，應將列表聲明為< extends T>類型，例如List< extends Integer>。然而，由於這種聲明方式的限制，你將無法向該列表添加任何元素。
相反，當作為消費者使用時，你需要一個列表可以使用T類型的元素，即你打算向其中添加T類型的元素。在這種情況下，應將列表聲明為< super T>類型，例如List< super Integer>。但是，這種聲明方式也帶來了限制，因為你無法保證從中讀取到的元素的具體類型。
值得注意的是，當一個列表既需要生產，又需要消費時，不能使用泛型通配符來聲明列表，例如List<>。這是因為這樣的聲明方式同時包含了生產者和消費者的需求，但在Java泛型中，這種雙重角色是不被允許的。
理解這一原則有助於更好地利用Java泛型，以更靈活和安全的方式設計和實現代碼。通過正確地應用PECS原則，可以有效避免許多常見的類型安全問題。