首页 > 代码库 > Java 线程池 +生产者消费者+MySQL读取300 万条数据

Java 线程池 +生产者消费者+MySQL读取300 万条数据

1.1需求

    数据库300 万条用户数据 ,遍历获取所有用户, 各种组合关联, 获取到一个新的json ,存到redis 上。

1.2 难点

  数据库比较多, 不可能单线程查询所有的数据到内存。

1.3解决办法

 多线程读取, 生产者 每次获取200 条数据, 消费者去消费。(这里 主要是根据MySQL分页去获取下一个200 条数据)

1.4 代码

1.4.1 调用方法

    /**
     * 线程启动
     */
    public void update() {
//redis操作类 HashRedisUtil redisUtil
= HashRedisUtil.getInstance(); //生产者消费者 ProducerConsumer pc = new ProducerConsumer(); //数据仓库 Storage s = pc.new Storage(); ExecutorService service = Executors.newCachedThreadPool(); //一个线程进行查询 Producer p = pc.new Producer(s,userMapper); service.submit(p); System.err.println("生产线程正在生产中。。。。。。。。。"); //是个线程进行修改 for(int i=0;i<10;i++){ System.err.println("消费线程"+i+"正在消费中。。。。。。。。。。"); service.submit(pc.new Consumer( redisUtil,userMapper,s)); } }

1.4.2 主要核心类

package com.ypp.thread;


import java.math.BigDecimal;
import java.util.Calendar;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.joda.time.LocalDateTime;

import com.alibaba.fastjson.JSONObject;
import com.ypp.constants.Constants;
import com.ypp.mapper.UserMapper;
import com.ypp.model.User;
import com.ypp.model.UserAlis;
import com.ypp.model.UserBaseModel;
import com.ypp.model.UserVip;
import com.ypp.util.HashRedisUtil;
import com.ypp.util.JsonUtils;
import com.ypp.util.PHPSerializer;


public class ProducerConsumer {
    private static Logger logger = Logger.getLogger(ProducerConsumer.class);
//这个page 是核心, 全局变量, 当生产者生产一次 ,获取200 个用户, 会把这个page++, 下次获取就是后一个200 条用户了
private static Integer page = 0; //消费者
public class Consumer implements Runnable { private HashRedisUtil redisUtil; private UserMapper userMapper; private Storage s = null; public Consumer(HashRedisUtil redisUtil, UserMapper userMapper, Storage s) { super(); this.redisUtil = redisUtil; this.userMapper = userMapper; this.s = s; } public void run() { try { while (true) { User users = s.pop(); long bbb = System.currentTimeMillis(); // 获取一个用户的粉丝列表 并存到redis try { fansUpdate(users.getToken(), users.getUserId(), redisUtil); } catch (Exception e1) { e1.printStackTrace(); } // 获取一个用户的关注列表, 并存到redis try { followUpdate(users.getToken(), users.getUserId(), redisUtil); } catch (Exception e) { e.printStackTrace(); } // 获取一个用户的黑名单, 并存到redis try { blackUpdate(users.getToken(), users.getUserId(), redisUtil); } catch (Exception e) { e.printStackTrace(); } // 用户基本信息 try { userbaseUpdate(users.getToken(), users.getUserId(), redisUtil); } catch (Exception e) { e.printStackTrace(); } long ccc = System.currentTimeMillis(); System.out.println("用户:" + users.getToken() + " 全部总共耗时:" + (ccc - bbb) + "毫秒"); Thread.sleep(500); } } catch (InterruptedException e) { e.printStackTrace(); } } public List<User> getUserInfo(Integer iThread) { return userMapper.findUserInfo((iThread - 1) * 200 + 1); } /** * 用户基本信息修改 * * @param token * @param myuserId * @param redisUtil * @throws Exception */ private void userbaseUpdate(String token, String myUserId, HashRedisUtil redisUtil) throws Exception { } /** * 更新一个用户的黑名单(原来的token改成userID) * * @param token * @param string * @param redisUtil * @throws Exception */ private void blackUpdate(String token, String myUserId, HashRedisUtil redisUtil) throws Exception { } /** * 获取一个用户的关注 * * @param token * @param string * @param redisUtil * @throws Exception */ private void followUpdate(String token, String myUserId, HashRedisUtil redisUtil) throws Exception { } /** * 获取一个用户的粉丝列表 * * @param token * @param userId * @param redisUtil * @throws Exception */ private void fansUpdate(String token, String myUserId, HashRedisUtil redisUtil) throws Exception { } //生产者 public class Producer implements Runnable { private Storage s = null; private UserMapper mapper ; public Producer( Storage s, UserMapper mapper) { this.s = s; this.mapper = mapper; } public void run() { try { while (true) { System.err.println("当前分页是:"+page+"****************************************"); List<User> list= mapper.findUserInfo(page); s.push(list); page++; } } catch (InterruptedException e1) { e1.printStackTrace(); } } }
//数据仓库
public class Storage { BlockingQueue<User> queues = new LinkedBlockingQueue<User>(200); /** * 生产 * * @param p * 产品 * @throws InterruptedException */ public void push(List<User> p) throws InterruptedException { for(User user:p){ queues.put(user); } } /** * 消费 * * @return 产品 * @throws InterruptedException */ public User pop() throws InterruptedException { return queues.take(); } } }

 

Java 线程池 +生产者消费者+MySQL读取300 万条数据