diff --git a/.classpath b/.classpath index fb50116..45cb5c9 100644 --- a/.classpath +++ b/.classpath @@ -1,6 +1,8 @@ + + diff --git a/.project b/.project index 6e26036..76d1107 100644 --- a/.project +++ b/.project @@ -1,6 +1,6 @@ - JerryDesignPattern + JerryMultiThread diff --git a/lib/mail.jar b/lib/mail.jar new file mode 100644 index 0000000..d1a4971 Binary files /dev/null and b/lib/mail.jar differ diff --git a/src/com/jerry/mail/Encrypt.java b/src/com/jerry/mail/Encrypt.java new file mode 100644 index 0000000..dbc15aa --- /dev/null +++ b/src/com/jerry/mail/Encrypt.java @@ -0,0 +1,71 @@ +package com.jerry.mail; + +import java.security.Key; +import java.security.SecureRandom; + +import javax.crypto.Cipher; +import javax.crypto.KeyGenerator; + +import sun.misc.BASE64Decoder; + +/** + * JAVA实现的DES加密解密算法 + * @author Jerry Wang + * + */ +public class Encrypt { + private Key key; + private byte[] byteMi = null; + private byte[] byteMing = null; + private String strM = ""; + + // 根据参数生成KEY + public void setKey(String strKey) { + try { + KeyGenerator generator = KeyGenerator.getInstance("DES"); + generator.init(new SecureRandom(strKey.getBytes())); + this.key = generator.generateKey(); + generator = null; + } catch (Exception e) { + e.printStackTrace(); + } + + } + + // 解密:以String密文输入,String明文输出 + public void setDesString(String strMi) { + BASE64Decoder base64De = new BASE64Decoder(); + try { + this.byteMi = base64De.decodeBuffer(strMi); + this.byteMing = this.getDesCode(byteMi); + this.strM = new String(byteMing, "UTF8"); + } catch (Exception e) { + e.printStackTrace(); + } finally { + base64De = null; + byteMing = null; + byteMi = null; + } + } + + // 解密以byte[]密文输入,以byte[]明文输出 + private byte[] getDesCode(byte[] byteD) { + Cipher cipher; + byte[] byteFina = null; + try { + cipher = Cipher.getInstance("DES"); + cipher.init(Cipher.DECRYPT_MODE, key); + byteFina = cipher.doFinal(byteD); + } catch (Exception e) { + e.printStackTrace(); + } finally { + cipher = null; + } + return byteFina; + } + + // 返回解密后的明文 + public String getStrM() { + return strM; + } +} \ No newline at end of file diff --git a/src/com/jerry/mail/SendMail.java b/src/com/jerry/mail/SendMail.java new file mode 100644 index 0000000..de10d33 --- /dev/null +++ b/src/com/jerry/mail/SendMail.java @@ -0,0 +1,343 @@ +/** + * 文件名称 : SendMail.java + * 项 目 名 : JavaMailWeb + * 包 名 : com.jerry.mail.model + * 版权所有 : 版权所有(C)2012-2013 + * 创建作者 : Jerry Wang + * 创建时间 : May 17, 2013 9:55:52 AM + * 电子邮件 : jerry002@126.com + * 当前版本 : v1.0 + */ +package com.jerry.mail; + +import java.util.ArrayList; +import java.util.Date; +import java.util.Iterator; +import java.util.Properties; +import java.util.StringTokenizer; + +import javax.activation.DataHandler; +import javax.activation.FileDataSource; +import javax.mail.Address; +import javax.mail.BodyPart; +import javax.mail.Message; +import javax.mail.MessagingException; +import javax.mail.Multipart; +import javax.mail.Session; +import javax.mail.Transport; +import javax.mail.internet.InternetAddress; +import javax.mail.internet.MimeBodyPart; +import javax.mail.internet.MimeMessage; +import javax.mail.internet.MimeMultipart; +import javax.mail.internet.MimeUtility; +/** + * SendMail.java + * 发送邮件类 + * @author Jerry Wang + * May 17, 2013 9:55:52 AM + */ +public class SendMail { + private static String SMTPHost = ""; // SMTP服务器 + private static String username = ""; // 登录SMTP服务器的帐号 + private static String password = ""; // 登录SMTP服务器的密码 + private static String from = ""; // 发件人邮箱 + + private Address[] to = null; // 收件人邮箱 + private String subject = ""; // 邮件标题 + private String content = ""; // 邮件内容 + private Address[] copyto = null;// 抄送邮件到 + private Session mailSession = null; + private Transport transport = null; + private ArrayList filename = new ArrayList(); // 附件文件名 + private static SendMail sendMail = null; + private final static String charset = "UTF-8"; + + /** + * @Name : SendMail + * @Description : 无参数构造方法 + * @param : + * @author : Jerry Wang + * @DateTime : May 17, 2013 9:59:47 AM + */ + private SendMail() {} + + /** + * @Name : getMailInstantiate + * @Description : 返回SendMail的对象 + * @param : @return + * @return : SendMail + * @author : Jerry Wang + * @DateTime : May 17, 2013 10:02:46 AM + */ + public static SendMail getInstance() { + if(null == sendMail) { + synchronized (SendMail.class) { + if(null == sendMail) { + init(); + sendMail = new SendMail(); + } + } + } + return sendMail; + } + + private synchronized static void init() { + SMTPHost = "smtp.126.com"; + username = "jerry002@126.com"; + from = username; + password = getPassword(); + } + + /** + * @Name : connect + * @Description : 连接SMTP邮件服务器 + * @param : + * @return : void + * @author : Jerry Wang + * @DateTime : May 17, 2013 10:16:13 AM + */ + public void connect() { + try { + if(transport == null || !transport.isConnected()) { + synchronized (this) { + if(transport == null || !transport.isConnected()) { + // 创建一个属性对象 + Properties props = new Properties(); + // 指定SMTP服务器 + props.put("mail.smtp.host", SMTPHost); + // 指定是否需要SMTP验证 + props.put("mail.smtp.auth", "true"); + // 创建一个授权验证对象 + SmtpPop3Auth auth = new SmtpPop3Auth(); + auth.setAccount(username, password); + // 创建一个Session对象 + mailSession = Session.getDefaultInstance(props, auth); + // 设置是否调试 + mailSession.setDebug(false); + // if (transport != null) + // 关闭连接 + // transport.close(); + // 创建一个Transport对象 + transport = mailSession.getTransport("smtp"); + // 连接SMTP服务器 + transport.connect(SMTPHost, username, password); + } + } + } + } catch (Exception e) { + e.printStackTrace(); + } + } + + /** + * @Name : close + * @Description : 关闭连接SMTP邮件服务器 + * @param : @return + * @return : void + * @author : Jerry Wang + * @DateTime : May 17, 2013 10:18:19 AM + */ + public void close() { + try { + if(transport.isConnected()) { + synchronized (SendMail.class) { + if(transport.isConnected()) { + transport.close(); + transport = null; + } + } + } + } catch (MessagingException e) { + e.printStackTrace(); + } + } + + /** + * @Name : send + * @Description : 发送邮件 + * @param : @return + * @return : String + * @author : Jerry Wang + * @DateTime : May 17, 2013 10:18:19 AM + */ + public synchronized boolean send() { + boolean result = false; + try { + // 连接smtp服务器 + sendMail.connect(); + // 创建一个MimeMessage 对象 + MimeMessage message = new MimeMessage(mailSession); + + // 指定发件人邮箱 + message.setFrom(new InternetAddress(from)); + // 指定收件人邮箱 + message.addRecipients(Message.RecipientType.TO, to); + if (!"".equals(copyto)) + // 指定抄送人邮箱 + message.addRecipients(Message.RecipientType.CC, copyto); + // 指定邮件主题 + message.setSubject(subject); + // 指定邮件发送日期 + message.setSentDate(new Date()); + // 指定邮件优先级 1：紧急 3：普通 5：缓慢 + message.setHeader("X-Priority", "3"); + message.saveChanges(); + // 判断附件是否为空 + if (!filename.isEmpty()) { + // 新建一个MimeMultipart对象用来存放多个BodyPart对象 + Multipart container = new MimeMultipart(); + // 新建一个存放信件内容的BodyPart对象 + BodyPart textBodyPart = new MimeBodyPart(); + // 给BodyPart对象设置内容和格式/编码方式 + textBodyPart.setContent(content, "text/html;charset="+charset); + // 将含有信件内容的BodyPart加入到MimeMultipart对象中 + container.addBodyPart(textBodyPart); + Iterator fileIterator = filename.iterator(); + while (fileIterator.hasNext()) {// 迭代所有附件 + String attachmentString = fileIterator.next(); + // 新建一个存放信件附件的BodyPart对象 + BodyPart fileBodyPart = new MimeBodyPart(); + // 将本地文件作为附件 + FileDataSource fds = new FileDataSource(attachmentString); + fileBodyPart.setDataHandler(new DataHandler(fds)); + // 处理邮件中附件文件名的中文问题 + String attachName = fds.getName(); + attachName = MimeUtility.encodeText(attachName); + // 设定附件文件名 + fileBodyPart.setFileName(attachName); + // 将附件的BodyPart对象加入到container中 + container.addBodyPart(fileBodyPart); + } + // 将container作为消息对象的内容 + message.setContent(container); + } else {// 没有附件的情况 + message.setContent(content, "text/html;charset="+charset); + } + // 发送邮件 + Transport.send(message, message.getAllRecipients()); + if (transport != null) + transport.close(); + result = true; + } catch (Exception e) { + e.printStackTrace(); + } finally { + sendMail.close(); + } + + return result; + } + + /** + * @Name : setContent + * @Description : 设置邮件内容 + * @param : @param content + * @return : void + * @author : Jerry Wang + * @DateTime : May 17, 2013 10:07:30 AM + */ + public void setContent(String content) { + this.content = content; + } + + /** + * @Name : setFilename + * @Description : 设置附件名称 + * @param : @param filename + * @return : void + * @author : Jerry Wang + * @DateTime : May 17, 2013 10:08:27 AM + */ + public void setFilename(ArrayList filename) { + try { + Iterator iterator = filename.iterator(); + ArrayList attachArrayList = new ArrayList(); + while (iterator.hasNext()) { + String attachment = iterator.next(); + // 解决文件名的中文问题 + attachment = MimeUtility.decodeText(attachment); + // 将文件路径中的'\'替换成'/' + attachment = attachment.replaceAll("\\\\", "/"); + attachArrayList.add(attachment); + } + this.filename = attachArrayList; + } catch (Exception ex) { + ex.printStackTrace(); + } + } + + /** + * @Name : setSubject + * @Description : 设置标题 + * @param : @param subject + * @return : void + * @author : Jerry Wang + * @DateTime : May 17, 2013 10:08:44 AM + */ + public void setSubject(String subject) { + try { + // 解决标题的中文问题 + subject = MimeUtility.encodeText(subject); + this.subject = subject; + } catch (Exception ex) { + ex.printStackTrace(); + } + } + + /** + * @Name : setTo + * @Description : 设置收件人邮箱 + * @param : @param toto + * @return : void + * @author : Jerry Wang + * @DateTime : May 17, 2013 10:09:09 AM + */ + public void setTo(String toto) { + try { + int i = 0; + StringTokenizer tokenizer = new StringTokenizer(toto, ";"); + to = new Address[tokenizer.countTokens()];// 动态的决定数组的长度 + while (tokenizer.hasMoreTokens()) { + String d = tokenizer.nextToken(); + + d = MimeUtility.encodeText(d); + to[i] = new InternetAddress(d);// 将字符串转换为整型 + + i++; + } + } catch (Exception e) { + e.printStackTrace(); + } + } + + /** + * @Name : setCopy_to + * @Description : 设置抄送 + * @param : @param copyTo + * @return : void + * @author : Jerry Wang + * @DateTime : May 17, 2013 10:09:38 AM + */ + public void setCopyto(String copyTo) { + try { + int i = 0; + StringTokenizer tokenizer = new StringTokenizer(copyTo, ";"); + copyto = new Address[tokenizer.countTokens()];// 动态的决定数组的长度 + while (tokenizer.hasMoreTokens()) { + String tolen = tokenizer.nextToken(); + tolen = MimeUtility.encodeText(tolen); + copyto[i] = new InternetAddress(tolen);// 将字符串转换为整型 + i++; + } + } catch (Exception e) { + e.printStackTrace(); + } + } + + public static String getPassword() { + Encrypt encrypt = new Encrypt(); + encrypt.setKey(username); + encrypt.setDesString("V2fwg3wuUDEI4xUTpYbzSA=="); + return encrypt.getStrM(); + } + +} diff --git a/src/com/jerry/mail/SendMailService.java b/src/com/jerry/mail/SendMailService.java new file mode 100644 index 0000000..202b862 --- /dev/null +++ b/src/com/jerry/mail/SendMailService.java @@ -0,0 +1,5 @@ +package com.jerry.mail; + +public interface SendMailService { + public void sendMail(String to, String subject, String content); +} diff --git a/src/com/jerry/mail/SendMailServiceImpl.java b/src/com/jerry/mail/SendMailServiceImpl.java new file mode 100644 index 0000000..162e4c2 --- /dev/null +++ b/src/com/jerry/mail/SendMailServiceImpl.java @@ -0,0 +1,31 @@ +package com.jerry.mail; + +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.ThreadPoolExecutor; + +public class SendMailServiceImpl implements SendMailService { + + @Override + public void sendMail(final String to, final String subject, final String content) { + ThreadPoolExecutor taskExecutor = new ScheduledThreadPoolExecutor(10); + taskExecutor.execute(new Runnable() { + + @Override + public void run() { + synchronized (SendMailServiceImpl.class) { + SendMail sendMail = SendMail.getInstance(); + sendMail.setSubject(subject); + sendMail.setTo(to); + sendMail.setContent(content); + long start = System.currentTimeMillis(); + sendMail.send(); + long end = System.currentTimeMillis(); + System.out.println("用时 ：" + (end - start) + "ms"); + System.out.println(Thread.currentThread() + "发送完成"); + } + } + }); + } + + +} diff --git a/src/com/jerry/mail/SmtpPop3Auth.java b/src/com/jerry/mail/SmtpPop3Auth.java new file mode 100644 index 0000000..cfacf63 --- /dev/null +++ b/src/com/jerry/mail/SmtpPop3Auth.java @@ -0,0 +1,20 @@ +package com.jerry.mail; + +import javax.mail.Authenticator; +import javax.mail.PasswordAuthentication; + +public class SmtpPop3Auth extends Authenticator { + public String user; + public String password; + + // 设置帐号信息 + public void setAccount(String user, String password) { + this.user = user; + this.password = password; + } + + // 取得PasswordAuthentication对象 + protected PasswordAuthentication getPasswordAuthentication() { + return new PasswordAuthentication(user, password); + } +} \ No newline at end of file diff --git a/src/com/jerry/performance/PerformanceTest.java b/src/com/jerry/performance/PerformanceTest.java new file mode 100644 index 0000000..160dee9 --- /dev/null +++ b/src/com/jerry/performance/PerformanceTest.java @@ -0,0 +1,187 @@ +package com.jerry.performance; + +import java.util.Random; +import java.util.concurrent.BrokenBarrierException; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantLock; + +/** + * Lock、synchronized、Atomic性能测试 + * @author Wangjiajun + * @Email wangjiajun@58.com + * + */ +public class PerformanceTest { + public static void test(int round, int threadNum) { + new SynchronizedTest("synchronized", round, threadNum).testTime(); + new LockTest("lock", round, threadNum).testTime(); + new AtomicTest("atomic", round, threadNum).testTime(); + } + + public static void main(String[] args) { + for(int i = 0; i < 5; i++) { + int round = 100000 * (i + 1); + int threadNum = 5 * (i + 1); + System.out.println("-------------------------"); + System.out.println("round:" + round + "; thread:" + threadNum); + PerformanceTest.test(round, threadNum); + + } + } +} + +class SynchronizedTest extends Template{ + + public SynchronizedTest(String id, int round, int threadNum) { + super(id, round, threadNum); + } + + @Override + synchronized void sumValue() { + super.countValue += super.preInit[index++%round]; + } + + @Override + synchronized long getValue() { + return super.countValue; + } + +} + +class LockTest extends Template { + ReentrantLock lock = new ReentrantLock(); + + public LockTest(String id, int round, int threadNum) { + super(id, round, threadNum); + } + + @Override + void sumValue() { + try { + lock.lock(); + super.countValue += super.preInit[index++%round]; + } finally { + lock.unlock(); + } + + } + + @Override + long getValue() { + try { + lock.lock(); + return super.countValue; + } finally { + lock.unlock(); + } + + } + +} + +class AtomicTest extends Template{ + + public AtomicTest(String id, int round, int threadNum) { + super(id, round, threadNum); + } + + @Override + void sumValue() { + super.countValueAtmoic.addAndGet(super.preInit[indexAtomic.get()%round]); + } + + @Override + long getValue() { + return super.countValueAtmoic.get(); + } + +} + +abstract class Template { + public String id; + public int round; + public int threadNum; + public long countValue; + public AtomicLong countValueAtmoic = new AtomicLong(0); + public int[] preInit; + public int index; + public AtomicInteger indexAtomic = new AtomicInteger(0); + Random r = new Random(47); + // 任务栅栏，同批任务，先到达wait的任务挂起，一直等到全部任务到达制定的wait地点后，才能全部唤醒，继续执行 + private CyclicBarrier cyclicBarrier = new CyclicBarrier(threadNum * 2 + 1); + + public Template(String id, int round, int threadNum) { + this.id = id; + this.round = round; + this.threadNum = threadNum; + preInit = new int[round]; + for (int i = 0; i < preInit.length; i++) { + preInit[i] = r.nextInt(100); + } + } + + abstract void sumValue(); + + /* + * 对long的操作是非原子的，原子操作只针对32位 long是64位， 底层操作的时候分2个32位读写，因此不是线程安全 + */ + abstract long getValue(); + + public void testTime() { + ExecutorService se = Executors.newCachedThreadPool(); + long start = System.nanoTime(); + // 同时开启2*ThreadNum个数的读写线程 + for (int i = 0; i < threadNum; i++) { + se.execute(new Runnable() { + public void run() { + for (int i = 0; i < round; i++) { + sumValue(); + } + + // 每个线程执行完同步方法后就等待 + try { + cyclicBarrier.await(); + } catch (InterruptedException e) { + e.printStackTrace(); + } catch (BrokenBarrierException e) { + e.printStackTrace(); + } + + } + }); + se.execute(new Runnable() { + public void run() { + + getValue(); + try { + // 每个线程执行完同步方法后就等待 + cyclicBarrier.await(); + } catch (InterruptedException e) { + e.printStackTrace(); + } catch (BrokenBarrierException e) { + e.printStackTrace(); + } + + } + }); + } + + try { + // 当前统计线程也wait,所以CyclicBarrier的初始值是threadNum*2+1 + cyclicBarrier.await(); + } catch (InterruptedException e) { + e.printStackTrace(); + } catch (BrokenBarrierException e) { + e.printStackTrace(); + } + // 所有线程执行完成之后，才会跑到这一步 + long duration = System.nanoTime() - start; + System.out.println(id + " = " + duration + "ms"); + + } + +} diff --git a/src/com/jerry/syncthreadpool/App.java b/src/com/jerry/syncthreadpool/App.java new file mode 100644 index 0000000..83ad5e1 --- /dev/null +++ b/src/com/jerry/syncthreadpool/App.java @@ -0,0 +1,30 @@ +package com.jerry.syncthreadpool; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; + +/** + * java线程池同步处理 + * @author Jerry Wang + * + */ +public class App { + + public static void main(String[] args) throws InterruptedException, ExecutionException { + + List lists = new ArrayList(); + for (int i = 0; i < 100000; i++) { + lists.add("jerry"+i); + } + + List> callables= SyncThreadPool.getSessionCallable(lists); + List> syncFnData = SyncThreadPool.syncFn(callables); + List results = new ArrayList(); + for (Future future : syncFnData) { + if(future.isDone()){ + results.add(future.get()) ; + } + } + } +} diff --git a/src/com/jerry/syncthreadpool/SessionCallable.java b/src/com/jerry/syncthreadpool/SessionCallable.java new file mode 100644 index 0000000..5309493 --- /dev/null +++ b/src/com/jerry/syncthreadpool/SessionCallable.java @@ -0,0 +1,7 @@ +package com.jerry.syncthreadpool; + +import java.util.concurrent.Callable; + +public abstract class SessionCallable implements Callable{ + +} \ No newline at end of file diff --git a/src/com/jerry/syncthreadpool/SyncThreadPool.java b/src/com/jerry/syncthreadpool/SyncThreadPool.java new file mode 100644 index 0000000..d05aa2e --- /dev/null +++ b/src/com/jerry/syncthreadpool/SyncThreadPool.java @@ -0,0 +1,42 @@ +package com.jerry.syncthreadpool; +import java.util.LinkedList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +public class SyncThreadPool { + + + public static List> getSessionCallable(List lists){ + List> callables = new LinkedList>(); + if(null != lists && lists.size() > 0){ + for (final String str : lists) { + callables.add(new SessionCallable() { + @Override + public String call() throws Exception { + String temp = str+"123"; + System.err.println(Thread.currentThread()+":"+str+":"+temp); + return temp; + } + }); + } + } + return callables; + } + + + public static List> syncFn(List> callables){ + try { + ExecutorService threadPool = Executors.newFixedThreadPool(5); + if(null!=callables&&callables.size()>0){ + List> Futures = threadPool.invokeAll(callables); + threadPool.shutdown(); + return Futures; + } + } catch (InterruptedException e) { + e.printStackTrace(); + } + return null; + } +} \ No newline at end of file diff --git a/src/com/jerry/thread/BlockingQueueCommunication.java b/src/com/jerry/thread/BlockingQueueCommunication.java new file mode 100644 index 0000000..619bdec --- /dev/null +++ b/src/com/jerry/thread/BlockingQueueCommunication.java @@ -0,0 +1,87 @@ +package com.jerry.thread; + +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; + +/** + * 阻塞队列通信 + * @author Jerry Wang + * + */ +public class BlockingQueueCommunication { + + /** + * @param args + */ + public static void main(String[] args) { + + final Business business = new Business(); + new Thread( + new Runnable() { + + @Override + public void run() { + + for(int i=1;i<=50;i++){ + business.sub(i); + } + + } + } + ).start(); + + for(int i=1;i<=50;i++){ + business.main(i); + } + + } + + static class Business { + + BlockingQueue queue1 = new ArrayBlockingQueue(1); + BlockingQueue queue2 = new ArrayBlockingQueue(1); + + { +// Collections.synchronizedMap(null); + try { + System.out.println("xxxxxdfsdsafdsa"); + queue2.put(1); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + + public void sub(int i){ + try { + queue1.put(1); + } catch (InterruptedException e) { + e.printStackTrace(); + } + for(int j=1;j<=10;j++){ + System.out.println("sub thread sequece of " + j + ",loop of " + i); + } + try { + queue2.take(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + + public void main(int i){ + try { + queue2.put(1); + } catch (InterruptedException e1) { + e1.printStackTrace(); + } + for(int j=1;j<=100;j++){ + System.out.println("main thread sequece of " + j + ",loop of " + i); + } + try { + queue1.take(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + } + +} diff --git a/src/com/jerry/thread/BlockingQueueTest.java b/src/com/jerry/thread/BlockingQueueTest.java new file mode 100644 index 0000000..aa0ae57 --- /dev/null +++ b/src/com/jerry/thread/BlockingQueueTest.java @@ -0,0 +1,51 @@ +package com.jerry.thread; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; + +/** + * 阻塞队列 BlockingQueue + * @author Jerry Wang + * + */ +public class BlockingQueueTest { + public static void main(String[] args) { + final BlockingQueue queue = new ArrayBlockingQueue(3); + for(int i=0;i<2;i++){ + new Thread(){ + public void run(){ + while(true){ + try { + Thread.sleep((long)(Math.random()*1000)); + System.out.println(Thread.currentThread().getName() + "准备放数据!"); + queue.put(1); + System.out.println(Thread.currentThread().getName() + "已经放了数据，" + + "队列目前有" + queue.size() + "个数据"); + } catch (InterruptedException e) { + e.printStackTrace(); + } + + } + } + + }.start(); + } + + new Thread(){ + public void run(){ + while(true){ + try { + //将此处的睡眠时间分别改为100和1000，观察运行结果 + Thread.sleep(1000); + System.out.println(Thread.currentThread().getName() + "准备取数据!"); + queue.take(); + System.out.println(Thread.currentThread().getName() + "已经取走数据，" + + "队列目前有" + queue.size() + "个数据"); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + } + + }.start(); + } +} diff --git a/src/com/jerry/thread/BoundedBuffer.java b/src/com/jerry/thread/BoundedBuffer.java new file mode 100644 index 0000000..35f2a0d --- /dev/null +++ b/src/com/jerry/thread/BoundedBuffer.java @@ -0,0 +1,53 @@ +package com.jerry.thread; + +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +/** + * 阻塞队列 + * @author Jerry Wang + * + */ +public class BoundedBuffer { + final Lock lock = new ReentrantLock(); + final Condition notFull = lock.newCondition(); + final Condition notEmpty = lock.newCondition(); + + final Object[] items = new Object[100]; + + int putptr, takeptr, count; + + public void put(Object x) throws InterruptedException { + lock.lock(); + + try { + while(count == items.length) + notFull.await(); + items[putptr] = x; + if(++putptr == items.length) + putptr = 0; + ++ count; + notEmpty.signal(); + } finally { + lock.unlock(); + } + } + + public Object take() throws InterruptedException { + try { + lock.lock(); + while(count == 0) + notEmpty.await(); + Object x = items[takeptr]; + if(++takeptr == items.length) + takeptr = 0; + --count; + notFull.signal(); + return x; + } finally { + lock.unlock(); + } + + } +} diff --git a/src/com/jerry/thread/CacheDemo.java b/src/com/jerry/thread/CacheDemo.java new file mode 100644 index 0000000..3ab94ef --- /dev/null +++ b/src/com/jerry/thread/CacheDemo.java @@ -0,0 +1,62 @@ +package com.jerry.thread; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +/** + * 模拟一个缓存器 + * + * @author Jerry Wang + * + */ +public class CacheDemo { + private Map cache = new HashMap(); + + /** + * 使用 synchronized 进行互斥 + * + * @param key + * @return + */ + // public synchronized Object getData(String key) { + // Object value = cache.get(key); + // if(value == null) { + // value = "xxxxxx"; // 实际是从数据库中取 + // } + // return value; + // } + + /** + * 使用读写锁 + * + * @param key + * @return + */ + private ReadWriteLock lock = new ReentrantReadWriteLock(); + + public Object get(String id) { + Object value = null; + lock.readLock().lock();// 首先开启读锁，从缓存中去取 + try { + value = cache.get(id); + if (value == null) { // 如果缓存中没有释放读锁，上写锁 + lock.readLock().unlock(); + lock.writeLock().lock(); + try { + if (value == null) { + value = "aaa"; // 此时可以去数据库中查找，这里简单的模拟一下 + } + } finally { + lock.writeLock().unlock(); // 释放写锁 + } + lock.readLock().lock(); // 然后再上读锁 + } + } finally { + lock.readLock().unlock(); // 最后释放读锁 + } + return value; + } + +} diff --git a/src/com/jerry/thread/CallableAndFuture.java b/src/com/jerry/thread/CallableAndFuture.java new file mode 100644 index 0000000..b2509e9 --- /dev/null +++ b/src/com/jerry/thread/CallableAndFuture.java @@ -0,0 +1,77 @@ +package com.jerry.thread; + +import java.util.Random; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +/** + * Callable与Future的应用 + * @author Jerry Wang + * + */ +public class CallableAndFuture { + public static void singleCallback() { + ExecutorService threadPool = Executors.newSingleThreadExecutor(); + Future future = + threadPool.submit(new Callable () { + + @Override + public String call() throws Exception { + Thread.sleep(2000); + return "hello"; + } + + }); + + System.out.println("等待结果"); + try { + System.out.println("拿到结果" + future.get(3,TimeUnit.SECONDS)); + } catch (InterruptedException e) { + e.printStackTrace(); + } catch (ExecutionException e) { + e.printStackTrace(); + } catch (TimeoutException e) { + e.printStackTrace(); + } + } + + public static void multiCallback() { + + ExecutorService threadPool2 = Executors.newFixedThreadPool(10); + ExecutorCompletionService executorCompletionService = new ExecutorCompletionService(threadPool2); + for(int i = 0; i < 10; i++) { + final int seq = i; + executorCompletionService.submit(new Callable() { + + @Override + public Integer call() throws Exception { + Thread.sleep(new Random().nextInt(5000)); + return seq; + } + + }); + } + + for(int i = 0; i < 10; i++) { + try { + System.out.println(executorCompletionService.take().get()); + } catch (InterruptedException e) { + e.printStackTrace(); + } catch (ExecutionException e) { + e.printStackTrace(); + } + } + + } + public static void main(String[] args) { +// CallableAndFuture.singleCallback(); + + CallableAndFuture.multiCallback(); + } +} diff --git a/src/com/jerry/thread/ConditionCommunication.java b/src/com/jerry/thread/ConditionCommunication.java new file mode 100644 index 0000000..7a428bb --- /dev/null +++ b/src/com/jerry/thread/ConditionCommunication.java @@ -0,0 +1,79 @@ +package com.jerry.thread; + +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +/** + * 使用condition实现线程同步通信技术 + * 子线程循环10次，接着主线程循环100次，接着又回到子线程循环10次， + * 接着再回到主线程又循环100次，如此循环50次 + * @author JerryWang + * + */ +public class ConditionCommunication { + public static void main(String[] args) { + final Business business = new Business(); + new Thread( + new Runnable() { + @Override + public void run() { + for(int i = 1; i <= 50; i++) { + business.sub(i); + } + } + } + ).start(); + + for(int i = 1; i <= 50; i++) { + business.main(i); + } + } + + static class Business{ + Lock lock = new ReentrantLock(); + Condition condition = lock.newCondition(); + private boolean bShouldSub = true; + public void sub(int i) { + lock.lock(); + try { + while(bShouldSub) { + try { + condition.await(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + for(int j = 1; j <= 10; j++) { + System.out.println("sub thread sequece of " + j + ", loop of " + i); + } + bShouldSub = false; + condition.signal(); + } finally { + lock.unlock(); + } + } + + public void main(int i) { + lock.lock(); + try { + while(!bShouldSub) { + try { + condition.await(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + for(int j = 1; j <= 100; j++) { + System.out.println("main thread sequece of " + j + ", loop of " + i); + } + bShouldSub = true; + condition.signal(); + } finally { + lock.unlock(); + } + } + } +} + + diff --git a/src/com/jerry/thread/CountdownLatchTest.java b/src/com/jerry/thread/CountdownLatchTest.java new file mode 100644 index 0000000..5621f2c --- /dev/null +++ b/src/com/jerry/thread/CountdownLatchTest.java @@ -0,0 +1,48 @@ +package com.jerry.thread; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * CountDownLatch同步工具 + * @author Jerry Wang + * + */ +public class CountdownLatchTest { + + public static void main(String[] args) { + ExecutorService service = Executors.newCachedThreadPool(); + final CountDownLatch cdOrder = new CountDownLatch(1); + final CountDownLatch cdAnswer = new CountDownLatch(3); + for(int i=0;i<3;i++){ + Runnable runnable = new Runnable(){ + public void run(){ + try { + System.out.println("线程" + Thread.currentThread().getName() + "正准备接受命令"); + cdOrder.await(); + System.out.println("线程" + Thread.currentThread().getName() + "已接受命令"); + Thread.sleep((long)(Math.random()*10000)); + System.out.println("线程" + Thread.currentThread().getName() + "回应命令处理结果"); + cdAnswer.countDown(); + } catch (Exception e) { + e.printStackTrace(); + } + } + }; + service.execute(runnable); + } + try { + Thread.sleep((long)(Math.random()*10000)); + + System.out.println("线程" + Thread.currentThread().getName() + "即将发布命令"); + cdOrder.countDown(); + System.out.println("线程" + Thread.currentThread().getName() + "已发送命令，正在等待结果"); + cdAnswer.await(); + System.out.println("线程" + Thread.currentThread().getName() + "已收到所有响应结果"); + } catch (Exception e) { + e.printStackTrace(); + } + service.shutdown(); + + } +} diff --git a/src/com/jerry/thread/CyclicBarrierTest.java b/src/com/jerry/thread/CyclicBarrierTest.java new file mode 100644 index 0000000..c765150 --- /dev/null +++ b/src/com/jerry/thread/CyclicBarrierTest.java @@ -0,0 +1,42 @@ +package com.jerry.thread; + +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * CyclicBarrier同步工具 + * @author Jerry Wang + * + */ +public class CyclicBarrierTest { + public static void main(String[] args) { + ExecutorService service = Executors.newCachedThreadPool(); + final CyclicBarrier cb = new CyclicBarrier(3); + for(int i=0;i<3;i++){ + Runnable runnable = new Runnable(){ + public void run(){ + try { + Thread.sleep((long)(Math.random()*10000)); + System.out.println("线程" + Thread.currentThread().getName() + + "即将到达集合地点1，当前已有" + (cb.getNumberWaiting()+1) + "个已经到达，" + (cb.getNumberWaiting()==2?"都到齐了，继续走啊":"正在等候")); + cb.await(); + + Thread.sleep((long)(Math.random()*10000)); + System.out.println("线程" + Thread.currentThread().getName() + + "即将到达集合地点2，当前已有" + (cb.getNumberWaiting()+1) + "个已经到达，" + (cb.getNumberWaiting()==2?"都到齐了，继续走啊":"正在等候")); + cb.await(); + Thread.sleep((long)(Math.random()*10000)); + System.out.println("线程" + Thread.currentThread().getName() + + "即将到达集合地点3，当前已有" + (cb.getNumberWaiting() + 1) + "个已经到达，" + (cb.getNumberWaiting()==2?"都到齐了，继续走啊":"正在等候")); + cb.await(); + } catch (Exception e) { + e.printStackTrace(); + } + } + }; + service.execute(runnable); + } + service.shutdown(); + } +} diff --git a/src/com/jerry/thread/ExchangerTest.java b/src/com/jerry/thread/ExchangerTest.java new file mode 100644 index 0000000..1d6f77a --- /dev/null +++ b/src/com/jerry/thread/ExchangerTest.java @@ -0,0 +1,49 @@ +package com.jerry.thread; +import java.util.concurrent.Exchanger; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * Exchanger同步工具 + * @author Jerry Wang + * + */ +public class ExchangerTest { + + public static void main(String[] args) { + ExecutorService service = Executors.newCachedThreadPool(); + final Exchanger exchanger = new Exchanger(); + service.execute(new Runnable(){ + public void run() { + try { + + String data1 = "zxx"; + System.out.println("线程" + Thread.currentThread().getName() + + "正在把数据" + data1 +"换出去"); + Thread.sleep((long)(Math.random()*10000)); + String data2 = (String)exchanger.exchange(data1); + System.out.println("线程" + Thread.currentThread().getName() + + "换回的数据为" + data2); + }catch(Exception e){ + + } + } + }); + service.execute(new Runnable(){ + public void run() { + try { + + String data1 = "lhm"; + System.out.println("线程" + Thread.currentThread().getName() + + "正在把数据" + data1 +"换出去"); + Thread.sleep((long)(Math.random()*10000)); + String data2 = (String)exchanger.exchange(data1); + System.out.println("线程" + Thread.currentThread().getName() + + "换回的数据为" + data2); + }catch(Exception e){ + + } + } + }); + } +} diff --git a/src/com/jerry/thread/ReadWriteLockTest.java b/src/com/jerry/thread/ReadWriteLockTest.java new file mode 100644 index 0000000..7896004 --- /dev/null +++ b/src/com/jerry/thread/ReadWriteLockTest.java @@ -0,0 +1,72 @@ +package com.jerry.thread; + +import java.util.Random; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +/** + * 读写锁 + * @author Jerry Wang + * + */ +public class ReadWriteLockTest { + public static void main(String[] args) { + final Queue queue = new Queue(); + for (int i = 0; i < 3; i++) { + new Thread() { + @Override + public void run() { + while(true) { + queue.get(); + } + } + }.start(); + + new Thread() { + @Override + public void run() { + while(true) { + queue.put(new Random().nextInt(10000)); + } + } + }.start(); + } + } +} + + +class Queue { + private Object data = null; + ReadWriteLock lock = new ReentrantReadWriteLock(); + public void get() { + lock.readLock().lock(); + try { + System.out.println(Thread.currentThread().getName() + " be ready to read data :" + data); + + Thread.sleep((long) (Math.random() * 1000)); + + System.out.println(Thread.currentThread().getName() + " have read data : " + data); + + } catch (InterruptedException e) { + e.printStackTrace(); + } finally { + lock.readLock().unlock(); + } + + } + + public void put(Object data) { + lock.writeLock().lock(); + try { + System.out.println(Thread.currentThread().getName() + " be ready to write data :" + data); + + Thread.sleep((long)(Math.random() * 1000)); + + System.out.println(Thread.currentThread().getName() + " have write data : " + data); + } catch (InterruptedException e) { + e.printStackTrace(); + } finally { + lock.writeLock().unlock(); + } + } +} diff --git a/src/com/jerry/thread/SemaphoreTest.java b/src/com/jerry/thread/SemaphoreTest.java new file mode 100644 index 0000000..4ae888b --- /dev/null +++ b/src/com/jerry/thread/SemaphoreTest.java @@ -0,0 +1,47 @@ +package com.jerry.thread; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Semaphore; + +/** + * Semaphore同步工具 + * @author Jerry Wang + * + */ +public class SemaphoreTest { + public static void main(String[] args) { + ExecutorService service = Executors.newCachedThreadPool(); + final Semaphore semaphore = new Semaphore(3); + for(int i = 0; i < 10; i++) { + Runnable runable = new Runnable() { + + @Override + public void run() { + try { + semaphore.acquire(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + + System.out.println("线程" + Thread.currentThread().getName() + "进入，" + + "当前已有" + (3 - semaphore.availablePermits() + "个并发")); + + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + e.printStackTrace(); + } + System.out.println("线程" + Thread.currentThread().getName() + "即将离开"); + + semaphore.release(); + + System.out.println("线程" + Thread.currentThread().getName() + "进入，" + + "当前已有" + (3 - semaphore.availablePermits() + "个并发")); + + } + }; + service.execute(runable); + } + } +} diff --git a/src/com/jerry/thread/TraditionalThreadSynchronized.java b/src/com/jerry/thread/TraditionalThreadSynchronized.java index d3ad116..95d6c4a 100644 --- a/src/com/jerry/thread/TraditionalThreadSynchronized.java +++ b/src/com/jerry/thread/TraditionalThreadSynchronized.java @@ -1,6 +1,10 @@ package com.jerry.thread; + +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + /** - * 传统方式实现线程互斥 + * 传统方式实现线程互斥(增加Lock) * @author Jerry Wang * */ @@ -52,12 +56,36 @@ class Outputer{ // } // } - public synchronized void output(String name) { +// public void output(String name) { +// int len = name.length(); +// synchronized(Outputer.class) { +// for(int i = 0; i < len; i++) { +// System.out.print(name.charAt(i)); +// } +// System.out.println(); +// } +// } + +// public synchronized void output(String name) { +// int len = name.length(); +// for(int i = 0; i < len; i++) { +// System.out.print(name.charAt(i)); +// } +// System.out.println(); +// } + + Lock lock = new ReentrantLock(); + public void output(String name) { int len = name.length(); - for(int i = 0; i < len; i++) { - System.out.print(name.charAt(i)); + lock.lock(); + try{ + for(int i = 0; i < len; i++) { + System.out.print(name.charAt(i)); + } + System.out.println(); + } finally { + lock.unlock(); } - System.out.println(); } } } diff --git a/src/com/jerry/threadpool/ThreadPool.java b/src/com/jerry/threadpool/ThreadPool.java new file mode 100644 index 0000000..6d68a35 --- /dev/null +++ b/src/com/jerry/threadpool/ThreadPool.java @@ -0,0 +1,105 @@ +package com.jerry.threadpool; + +import java.util.LinkedList; + +/** + * Java线程池工具类 + * @author Jerry Wang + * + */ +public class ThreadPool extends ThreadGroup { + private boolean isClosed = false; //线程池是否关闭 + private LinkedList workQueue; //工作队列 + private static int threadPoolID = 1; //线程池的id + public ThreadPool(int poolSize) { //poolSize 表示线程池中的工作线程的数量 + + super(threadPoolID + ""); //指定ThreadGroup的名称 + setDaemon(true); //继承到的方法，设置是否守护线程池 + workQueue = new LinkedList(); //创建工作队列 + for(int i = 0; i < poolSize; i++) { + new WorkThread(i).start(); //创建并启动工作线程,线程池数量是多少就创建多少个工作线程 + } + } + + /** 向工作队列中加入一个新任务,由工作线程去执行该任务*/ + public synchronized void execute(Runnable task) { + if(isClosed) { + throw new IllegalStateException(); + } + if(task != null) { + workQueue.add(task);//向队列中加入一个任务 + notify(); //唤醒一个正在getTask()方法中待任务的工作线程 + } + } + + /** 从工作队列中取出一个任务,工作线程会调用此方法*/ + private synchronized Runnable getTask(int threadid) throws InterruptedException { + while(workQueue.size() == 0) { + if(isClosed) return null; + System.out.println("工作线程"+threadid+"等待任务..."); + wait(); //如果工作队列中没有任务,就等待任务 + } + System.out.println("工作线程"+threadid+"开始执行任务..."); + return (Runnable) workQueue.removeFirst(); //反回队列中第一个元素,并从队列中删除 + } + + /** 关闭线程池 */ + public synchronized void closePool() { + if(! isClosed) { + waitFinish(); //等待工作线程执行完毕 + isClosed = true; + workQueue.clear(); //清空工作队列 + interrupt(); //中断线程池中的所有的工作线程,此方法继承自ThreadGroup类 + } + } + + /** 等待工作线程把所有任务执行完毕*/ + public void waitFinish() { + synchronized (this) { + isClosed = true; + notifyAll(); //唤醒所有还在getTask()方法中等待任务的工作线程 + } + Thread[] threads = new Thread[activeCount()]; //activeCount() 返回该线程组中活动线程的估计值。 + int count = enumerate(threads); //enumerate()方法继承自ThreadGroup类，根据活动线程的估计值获得线程组中当前所有活动的工作线程 + for(int i =0; i < count; i++) { //等待所有工作线程结束 + try { + threads[i].join(); //等待工作线程结束 + }catch(InterruptedException ex) { + ex.printStackTrace(); + } + } + } + + /** + * 内部类,工作线程,负责从工作队列中取出任务,并执行 + * @author sunnylocus + */ + private class WorkThread extends Thread { + private int id; + public WorkThread(int id) { + //父类构造方法,将线程加入到当前ThreadPool线程组中 + super(ThreadPool.this,id+""); + this.id =id; + } + + @Override + public void run() { + while(! isInterrupted()) { //isInterrupted()方法继承自Thread类，判断线程是否被中断 + Runnable task = null; + try { + task = getTask(id); //取出任务 + }catch(InterruptedException ex) { + ex.printStackTrace(); + } + //如果getTask()返回null或者线程执行getTask()时被中断，则结束此线程 + if(task == null) return; + + try { + task.run(); //运行任务 + }catch(Throwable t) { + t.printStackTrace(); + } + }// end while + }// end run + }// end workThread +} \ No newline at end of file diff --git a/src/com/jerry/threadpool/ThreadPoolTest.java b/src/com/jerry/threadpool/ThreadPoolTest.java new file mode 100644 index 0000000..40e238d --- /dev/null +++ b/src/com/jerry/threadpool/ThreadPoolTest.java @@ -0,0 +1,32 @@ +package com.jerry.threadpool; + + +/** + * Java线程池工具类测试 + * @author Jerry Wang + * + */ +public class ThreadPoolTest { + + public static void main(String[] args) throws InterruptedException { + ThreadPool threadPool = new ThreadPool(3); //创建一个有个3工作线程的线程池 + Thread.sleep(500); //休眠500毫秒,以便让线程池中的工作线程全部运行 + //运行任务 + for (int i = 0; i <=5 ; i++) { //创建6个任务 + threadPool.execute(createTask(i)); + } + threadPool.waitFinish(); //等待所有任务执行完毕 + threadPool.closePool(); //关闭线程池 + + } + + private static Runnable createTask(final int taskID) { + return new Runnable() { + public void run() { + // System.out.println("Task" + taskID + "开始"); + System.out.println("Hello world"); + // System.out.println("Task" + taskID + "结束"); + } + }; + } +} diff --git a/src/com/jerry/wait/ThreadWait.java b/src/com/jerry/wait/ThreadWait.java new file mode 100644 index 0000000..dfeb03b --- /dev/null +++ b/src/com/jerry/wait/ThreadWait.java @@ -0,0 +1,207 @@ +package com.jerry.wait; + +/** + * 线程等待的几种方法 + */ +import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.Date; +import java.util.List; +import java.util.concurrent.BrokenBarrierException; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +public class ThreadWait { + final static SimpleDateFormat sdf = new SimpleDateFormat( + "yyyy-MM-dd HH:mm:ss"); + + public static void main(String[] args) throws Exception { + join(); + + } + + private void doSomeWork() { + try { + Thread.sleep((long) (Math.random() * 10000)); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + + public static void doSuperWork() { + System.out.println("Super Worker begin at " + sdf.format(new Date())); + try { + Thread.sleep((long) (Math.random() * 10000)); + } catch (InterruptedException e) { + e.printStackTrace(); + } + System.out.println("Super Worker end at " + sdf.format(new Date())); + } + + static class JoinWorker extends Thread { + String workerName; + + public JoinWorker(String workerName) { + this.workerName = workerName; + } + + public void run() { + System.out.println("Sub Worker " + workerName + + " do work begin at " + sdf.format(new Date())); + new ThreadWait().doSomeWork();// 做实际工作 + System.out.println("Sub Worker " + workerName + + " do work complete at " + sdf.format(new Date())); + } + } + + /** + * 使用线程自带的join方法，将子线程加到主线程中 主线程需要等待子线程完成才继续执行 + * + * @throws InterruptedException + */ + public static void join() throws InterruptedException { + System.out.println("=========Test with join====="); + JoinWorker worker1 = new JoinWorker("worker1"); + JoinWorker worker2 = new JoinWorker("worker2"); + worker1.start(); + worker2.start(); + worker1.join(); + worker2.join(); + doSuperWork(); + } + + /** + * 使用CountDownLatch，每个线程调用其countDown方法使计数器-1， + * 主线程调用await方法阻塞等待，直到CountDownLatch计数器为0时继续执行 + * + * @throws InterruptedException + */ + public static void countDownLatch() throws InterruptedException { + System.out.println("=========Test with CountDownLatch====="); + CountDownLatch latch = new CountDownLatch(2); + CountDownLatchWorker worker1 = new CountDownLatchWorker("worker1",latch); + CountDownLatchWorker worker2 = new CountDownLatchWorker("worker2",latch); + worker1.start(); + worker2.start(); + // 主线程阻塞等待 + latch.await(); + doSuperWork(); + } + + /** + * CyclicBarrier类似于CountDownLatch也是个计数器， 不同的是CyclicBarrier的await() + * 方法没被调用一次，计数便会减少1，并阻塞住当前线程。当计数减至0时，阻塞解除，所有在此 CyclicBarrier 上面阻塞的线程开始运行。 + * 在这之后，如果再次调用 await() 方法，计数就又会变成 N-1，新一轮重新开始 + * CyclicBarrier初始时还可带一个Runnable的参数， + * 此Runnable任务在CyclicBarrier的数目达到后，所有其它线程被唤醒前被执行。 + */ + public static void cyclicBarrier() throws InterruptedException, + BrokenBarrierException { + System.out.println("=========Test with CyclicBarrier====="); + CyclicBarrier cb = new CyclicBarrier(2, new Runnable() { + // 将主线程业务放到CyclicBarrier构造方法中，所有线程都到达Barrier时执行 + @SuppressWarnings("static-access") + public void run() { + new ThreadWait().doSuperWork(); + } + });// 设定需要等待两个线程 + ExecutorService executor = Executors.newFixedThreadPool(2); + CyclicBarrierWorker worker1 = new CyclicBarrierWorker("worker1", cb); + CyclicBarrierWorker worker2 = new CyclicBarrierWorker("worker2", cb); + executor.execute(worker1); + executor.execute(worker2); + executor.shutdown(); + } + + /** + * 使用ExecutorService的invokeAll方法调研callable集合，批量执行多个线程 + * 在invokeAll方法结束之后，再执行主线程其他业务逻辑 + * + * @throws InterruptedException + */ + public static void callable() throws InterruptedException { + System.out.println("=========Test with Callable====="); + List> callList = new ArrayList>(); + ExecutorService exec = Executors.newFixedThreadPool(2); + // 采用匿名内部类实现 + callList.add(new Callable() { + public Integer call() throws Exception { + System.out.println("Sub Worker worker1 do work begin at " + + sdf.format(new Date())); + new ThreadWait().doSomeWork();// 做实际工作 + System.out.println("Sub Worker worker1 do work complete at " + + sdf.format(new Date())); + return 0; + } + }); + callList.add(new Callable() { + public Integer call() throws Exception { + System.out.println("Sub Worker worker2 do work begin at " + + sdf.format(new Date())); + new ThreadWait().doSomeWork();// 做实际工作 + System.out.println("Sub Worker worker2 do work complete at " + + sdf.format(new Date())); + return 0; + } + }); + exec.invokeAll(callList); + exec.shutdown(); + doSuperWork(); + + } + + static class CountDownLatchWorker extends Thread { + String workerName; + + CountDownLatch latch; + + public CountDownLatchWorker(String workerName, CountDownLatch latch) { + this.workerName = workerName; + this.latch = latch; + } + + public void run() { + System.out.println("Sub Worker " + workerName + + " do work begin at " + sdf.format(new Date())); + new ThreadWait().doSomeWork();// 做实际工作 + System.out.println("Sub Worker " + workerName + + " do work complete at " + sdf.format(new Date())); + latch.countDown();// 完成之后，计数器减一 + + } + } + + static class CyclicBarrierWorker extends Thread { + String workerName; + + CyclicBarrier cb; + + public CyclicBarrierWorker(String workerName, CyclicBarrier cb) { + super(); + this.workerName = workerName; + this.cb = cb; + } + + public void run() { + System.out.println("Sub Worker " + workerName + + " do work begin at " + sdf.format(new Date())); + new ThreadWait().doSomeWork();// 做实际工作 + System.out.println("Sub Worker " + workerName + + " do work complete at " + sdf.format(new Date())); + try { + // 等待其他未完成线程 + cb.await(); + } catch (InterruptedException e) { + e.printStackTrace(); + } catch (BrokenBarrierException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + + } + } + +} \ No newline at end of file