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

1.4.2 主要核心类

<span style="color: #0000ff">package</span><span style="color: #000000"> com.ypp.thread;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> java.math.BigDecimal;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> java.util.Calendar;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> java.util.HashMap;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> java.util.List;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> java.util.Map;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> java.util.Set;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> java.util.concurrent.BlockingQueue;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> java.util.concurrent.LinkedBlockingQueue;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> org.apache.commons.lang.StringUtils;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> org.apache.log4j.Logger;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> org.joda.time.LocalDateTime;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> com.alibaba.fastjson.JSONObject;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> com.ypp.constants.Constants;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> com.ypp.mapper.UserMapper;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> com.ypp.model.User;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> com.ypp.model.UserAlis;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> com.ypp.model.UserBaseModel;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> com.ypp.model.UserVip;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> com.ypp.util.HashRedisUtil;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> com.ypp.util.JsonUtils;
</span><span style="color: #0000ff">import</span><span style="color: #000000"> com.ypp.util.PHPSerializer;
</span>
<span style="color: #0000ff">public</span> <span style="color: #0000ff">class</span><span style="color: #000000"> ProducerConsumer {
    </span><span style="color: #0000ff">private</span> <span style="color: #0000ff">static</span> Logger logger = Logger.getLogger(ProducerConsumer.<span style="color: #0000ff">class</span><span style="color: #000000">);<br>    //这个page 是核心， 全局变量， 当生产者生产一次 ，获取200 个用户， 会把这个page++, 下次获取就是后一个200 条用户了
    </span><span style="color: #0000ff">private</span> <span style="color: #0000ff">static</span> Integer page = 0<span style="color: #000000">;
  
   //消费者<br>    
    </span><span style="color: #0000ff">public</span> <span style="color: #0000ff">class</span> Consumer <span style="color: #0000ff">implements</span><span style="color: #000000"> Runnable {
        
        </span><span style="color: #0000ff">private</span><span style="color: #000000"> HashRedisUtil redisUtil;
        </span><span style="color: #0000ff">private</span><span style="color: #000000"> UserMapper userMapper;
        </span><span style="color: #0000ff">private</span> Storage s = <span style="color: #0000ff">null</span><span style="color: #000000">;
        </span><span style="color: #0000ff">public</span><span style="color: #000000"> Consumer(HashRedisUtil redisUtil, UserMapper userMapper, Storage s) {
            </span><span style="color: #0000ff">super</span><span style="color: #000000">();
            </span><span style="color: #0000ff">this</span>.redisUtil =<span style="color: #000000"> redisUtil;
            </span><span style="color: #0000ff">this</span>.userMapper =<span style="color: #000000"> userMapper;
            </span><span style="color: #0000ff">this</span>.s =<span style="color: #000000"> s;
        }
        </span><span style="color: #0000ff">public</span> <span style="color: #0000ff">void</span><span style="color: #000000"> run() {
            </span><span style="color: #0000ff">try</span><span style="color: #000000"> {
                </span><span style="color: #0000ff">while</span> (<span style="color: #0000ff">true</span><span style="color: #000000">) {
                    User users </span>=<span style="color: #000000"> s.pop();
                    
                    </span><span style="color: #0000ff">long</span> bbb =<span style="color: #000000"> System.currentTimeMillis();
                    </span><span style="color: #008000">//</span><span style="color: #008000"> 获取一个用户的粉丝列表 并存到redis</span>
                    <span style="color: #0000ff">try</span><span style="color: #000000"> {
                        fansUpdate(users.getToken(), users.getUserId(), redisUtil);
                    } </span><span style="color: #0000ff">catch</span><span style="color: #000000"> (Exception e1) {
                        e1.printStackTrace();
                    }
                    </span><span style="color: #008000">//</span><span style="color: #008000"> 获取一个用户的关注列表， 并存到redis</span>
                    <span style="color: #0000ff">try</span><span style="color: #000000"> {
                        followUpdate(users.getToken(), users.getUserId(), redisUtil);
                    } </span><span style="color: #0000ff">catch</span><span style="color: #000000"> (Exception e) {
                        e.printStackTrace();
                    }
                    </span><span style="color: #008000">//</span><span style="color: #008000"> 获取一个用户的黑名单， 并存到redis</span>
                    <span style="color: #0000ff">try</span><span style="color: #000000"> {
                        blackUpdate(users.getToken(), users.getUserId(), redisUtil);
                    } </span><span style="color: #0000ff">catch</span><span style="color: #000000"> (Exception e) {
                        e.printStackTrace();
                    }
                    </span><span style="color: #008000">//</span><span style="color: #008000"> 用户基本信息</span>
                    <span style="color: #0000ff">try</span><span style="color: #000000"> {
                        userbaseUpdate(users.getToken(), users.getUserId(), redisUtil);
                    } </span><span style="color: #0000ff">catch</span><span style="color: #000000"> (Exception e) {
                        e.printStackTrace();
                    }
                    </span><span style="color: #0000ff">long</span> ccc =<span style="color: #000000"> System.currentTimeMillis();
                    System.out.println(</span>"用户：" + users.getToken() + " 全部总共耗时：" + (ccc - bbb) + "毫秒"<span style="color: #000000">);
                    Thread.sleep(</span>500<span style="color: #000000">);
                }
            } </span><span style="color: #0000ff">catch</span><span style="color: #000000"> (InterruptedException e) {
                e.printStackTrace();
            }
        }
        
        </span><span style="color: #0000ff">public</span> List<User><span style="color: #000000"> getUserInfo(Integer iThread) {
            </span><span style="color: #0000ff">return</span> userMapper.findUserInfo((iThread - 1) * 200 + 1<span style="color: #000000">);
        }
    </span><span style="color: #008000">/**</span><span style="color: #008000">
         * 用户基本信息修改
         * 
         * </span><span style="color: #808080">@param</span><span style="color: #008000"> token
         * </span><span style="color: #808080">@param</span><span style="color: #008000"> myuserId
         * </span><span style="color: #808080">@param</span><span style="color: #008000"> redisUtil
         * </span><span style="color: #808080">@throws</span><span style="color: #008000"> Exception
         </span><span style="color: #008000">*/</span>
        <span style="color: #0000ff">private</span> <span style="color: #0000ff">void</span> userbaseUpdate(String token, String myUserId, HashRedisUtil redisUtil) <span style="color: #0000ff">throws</span><span style="color: #000000"> Exception {
          </span>
<span style="color: #000000">
        }
        </span><span style="color: #008000">/**</span><span style="color: #008000">
         * 更新一个用户的黑名单(原来的token改成userID)
         * 
         * </span><span style="color: #808080">@param</span><span style="color: #008000"> token
         * </span><span style="color: #808080">@param</span><span style="color: #008000"> string
         * </span><span style="color: #808080">@param</span><span style="color: #008000"> redisUtil
         * </span><span style="color: #808080">@throws</span><span style="color: #008000"> Exception
         </span><span style="color: #008000">*/</span>
        <span style="color: #0000ff">private</span> <span style="color: #0000ff">void</span> blackUpdate(String token, String myUserId, HashRedisUtil redisUtil) <span style="color: #0000ff">throws</span><span style="color: #000000"> Exception {
           </span><span style="color: #000000">
        }
        </span><span style="color: #008000">/**</span><span style="color: #008000">
         * 获取一个用户的关注
         * 
         * </span><span style="color: #808080">@param</span><span style="color: #008000"> token
         * </span><span style="color: #808080">@param</span><span style="color: #008000"> string
         * </span><span style="color: #808080">@param</span><span style="color: #008000"> redisUtil
         * </span><span style="color: #808080">@throws</span><span style="color: #008000"> Exception
         </span><span style="color: #008000">*/</span>
        <span style="color: #0000ff">private</span> <span style="color: #0000ff">void</span> followUpdate(String token, String myUserId, HashRedisUtil redisUtil) <span style="color: #0000ff">throws</span><span style="color: #000000"> Exception {
          </span><span style="color: #000000">
        }
        </span><span style="color: #008000">/**</span><span style="color: #008000">
         * 获取一个用户的粉丝列表
         * 
         * </span><span style="color: #808080">@param</span><span style="color: #008000"> token
         * </span><span style="color: #808080">@param</span><span style="color: #008000"> userId
         * </span><span style="color: #808080">@param</span><span style="color: #008000"> redisUtil
         * </span><span style="color: #808080">@throws</span><span style="color: #008000"> Exception
         </span><span style="color: #008000">*/</span>
        <span style="color: #0000ff">private</span> <span style="color: #0000ff">void</span> fansUpdate(String token, String myUserId, HashRedisUtil redisUtil) <span style="color: #0000ff">throws</span><span style="color: #000000"> Exception {</span><span style="color: #000000">
        }
  </span><span style="color: #000000">
//生产者
    </span><span style="color: #0000ff">public</span> <span style="color: #0000ff">class</span> Producer <span style="color: #0000ff">implements</span><span style="color: #000000"> Runnable {
        </span><span style="color: #0000ff">private</span> Storage s = <span style="color: #0000ff">null</span><span style="color: #000000">;
        </span><span style="color: #0000ff">private</span><span style="color: #000000"> UserMapper mapper ;
        </span><span style="color: #0000ff">public</span><span style="color: #000000"> Producer( Storage s, UserMapper mapper) {
            
            </span><span style="color: #0000ff">this</span>.s =<span style="color: #000000"> s;
            </span><span style="color: #0000ff">this</span>.mapper =<span style="color: #000000"> mapper;
        }
        </span><span style="color: #0000ff">public</span> <span style="color: #0000ff">void</span><span style="color: #000000"> run() {
            </span><span style="color: #0000ff">try</span><span style="color: #000000"> {
                
                </span><span style="color: #0000ff">while</span> (<span style="color: #0000ff">true</span><span style="color: #000000">) {
                    
                    System.err.println(</span>"当前分页是："+page+"****************************************"<span style="color: #000000">);
                     List</span><User> list=<span style="color: #000000"> mapper.findUserInfo(page);
                    s.push(list);
                    page</span>++<span style="color: #000000">;
                }
            } </span><span style="color: #0000ff">catch</span><span style="color: #000000"> (InterruptedException e1) {
                e1.printStackTrace();
            }
        }
    }
<br>//数据仓库
    </span><span style="color: #0000ff">public</span> <span style="color: #0000ff">class</span><span style="color: #000000"> Storage {
        BlockingQueue</span><User> queues = <span style="color: #0000ff">new</span> LinkedBlockingQueue<User>(200<span style="color: #000000">);
        </span><span style="color: #008000">/**</span><span style="color: #008000">
         * 生产
         * 
         * </span><span style="color: #808080">@param</span><span style="color: #008000"> p
         *            产品
         * </span><span style="color: #808080">@throws</span><span style="color: #008000"> InterruptedException
         </span><span style="color: #008000">*/</span>
        <span style="color: #0000ff">public</span> <span style="color: #0000ff">void</span> push(List<User> p) <span style="color: #0000ff">throws</span><span style="color: #000000"> InterruptedException {
            </span><span style="color: #0000ff">for</span><span style="color: #000000">(User user:p){
                queues.put(user);
            }
        }
        </span><span style="color: #008000">/**</span><span style="color: #008000">
         * 消费
         * 
         * </span><span style="color: #808080">@return</span><span style="color: #008000"> 产品
         * </span><span style="color: #808080">@throws</span><span style="color: #008000"> InterruptedException
         </span><span style="color: #008000">*/</span>
        <span style="color: #0000ff">public</span> User pop() <span style="color: #0000ff">throws</span><span style="color: #000000"> InterruptedException {
            </span><span style="color: #0000ff">return</span><span style="color: #000000"> queues.take();
        }
    }
}</span>

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

标签：用户   ase   mode   cache   代码   log   shm   nbsp   inf