目标：为提高数据库表更新（update）效率，使用多线程更新。其实这里也可以考虑另一种方法批量更新，不过如果更新失败了，同一事务（transaction）中的其他更新语句就会回滚，比较麻烦，所在还是简单点用多线程去处理。
困难点：开始是想把需要更新的数据等分到线程中去处理，不过搞了一段时间都没成功，主要是没有什么办法把动态参数从线程外传进线程内。后来换了个思路，每个线程去拿一条数据去更新，拿数据时同步处理。这样之后就可以了。
例子（部分外部类没有写上去，所以只能参考，要运行需要改改）：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 |
import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; import javax.annotation.Resource; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; public class BaiduBaikeJob { static final Logger logger = LoggerFactory .getLogger(BaiduBaikeJob.class); @Autowired private MongoLockService mongoLockService; @Autowired private WeiboPersonService weiboPersonService; @Resource(name = "globalProperties") private PropertyPlaceholderConfigurer config; final int THREAD_COUNT = 10; AtomicInteger updCounter = new AtomicInteger(0); AtomicInteger next = new AtomicInteger(0); private CountDownLatch threadCompletedCounter = new CountDownLatch(THREAD_COUNT); /** * 每隔 1小时，获得百度百科里的人物信息 */ public void baiduBaikeRun() { if ("true".equalsIgnoreCase((String) config.getProperty("runBaiduBaikeJob", "false")) && mongoLockService.getLock(MongoLockService.LOCK_BAIDU_BAIKE)) { List<WeiboPerson> unBaikeFills = null; try { unBaikeFills = weiboPersonService.findUnBaikeFills(2000); if (unBaikeFills != null && !unBaikeFills.isEmpty()) { List<WeiboPerson> weiboPersonList = new ArrayList<WeiboPerson>(); for (final WeiboPerson person : unBaikeFills) { // 获取百度百科信息 WeiboPerson weiboPerson = BaiduBaikeUtils.getBaiduBaikeInfo(person.getName()); weiboPerson.setId(person.getId()); weiboPerson.setName(person.getName()); weiboPerson.setTenantId(person.getTenantId()); weiboPerson.setGroupId(person.getGroupId()); // 更新状态 if (!StringUtils.isEmptyString(weiboPerson.getDescription())) { weiboPerson.setFlagBaike(WeiboPerson.BAIKE_SUCCESS); } else { weiboPerson.setFlagBaike(WeiboPerson.BAIKE_FAIL); } weiboPerson.setLastUpdDate(new Date()); weiboPersonList.add(weiboPerson); } // 首次更新或过期更新 baiduBaikeRunInMultiThreads(weiboPersonList); } } catch (Exception e) { logger.error("update baidu baike error，" + e.getClass().getName() + ": " + e.getMessage()); } finally { try { mongoLockService.releaseLock(MongoLockService.LOCK_BAIDU_BAIKE); } catch (Exception e) { logger.error("release lock error ", e); } } } } /** * 多线程处理：更新table */ private void baiduBaikeRunInMultiThreads(final List<WeiboPerson> weiboPersonList) { ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT); for (int i = 0; i < THREAD_COUNT; i++) { executor.submit(new Runnable() { public void run() { WeiboPerson weiboPerson = getNext(weiboPersonList); while (null != weiboPerson) { try { // 首次更新或过期更新 weiboPersonService.update(weiboPerson); } catch (Exception e) { logger.error("table weiboPerson update error ", e); } updCounter.incrementAndGet(); // System.out.println("Thread:" + Thread.currentThread().getName() + ", counter:" + updCounter + ", name:" + weiboPerson.getName()); weiboPerson = getNext(weiboPersonList); } threadCompletedCounter.countDown(); } }); } closeThreadPool(executor); } /** * 同步处理：获取需要更新的一条微博人物 */ private synchronized WeiboPerson getNext(List<WeiboPerson> weiboPersonList){ if(next.intValue()>=weiboPersonList.size()) return null; next.incrementAndGet(); return weiboPersonList.get(next.intValue()-1); } /** * 关闭线程池 */ private void closeThreadPool(final ExecutorService executor) { try { threadCompletedCounter.await(); executor.shutdown(); } catch (InterruptedException e) { e.printStackTrace(); } } public static void main(String[] args) { // mock List<WeiboPerson> weiboPersonList = new ArrayList<WeiboPerson>(); int i = 1; while (i <= 10) { WeiboPerson weiboPerson = new WeiboPerson(); weiboPerson.setName("a" + i); weiboPersonList.add(weiboPerson); i++; } // test multi-thread BaiduBaikeJob baiduBaikeJob = new BaiduBaikeJob(); baiduBaikeJob.baiduBaikeRunInMultiThreads(weiboPersonList); } } |