现在有这样一个任务，有一份手机号列表（20W），有一份话单的列表（10W），要统计哪些手机号没有出现在话单中，哪些手机号在话单中出现了不止一次。
想到的最直接的方式，就是两层循环去遍历，虽然此方法比较笨，但目前还没有想出更好的办法。
一开始使用单线程来处理，代码是随手写的并没有进行重构，只是做一个简单的说明：
package tool;
import java.util.List;
public class SingleThread
{
public static void main(String[] args)
{
SingleThread st = new SingleThread();
String userIdPath = "D:\\shell\\store_bak\\tool\\userid.txt";
List<String> userIds = Util.readUserId(userIdPath);
List<String> cdrItems = Util.readCdrItem();
st.process(userIds, cdrItems);
}
/**
*
* @param userIds
* @param cdrItems
*/
private void process(List<String> userIds, List<String> cdrItems)
{
long startTime = System.currentTimeMillis();
int count = 0;
for (String key : userIds)
{
String[] uninKeys = key.split("\\s+");
count = 0;
for (String cdr : cdrItems)
{
if (cdr.contains("|" + uninKeys[0] + "|")
&& cdr.contains("|" + uninKeys[1] + "|"))
{
count++;
}
}
}
System.out.println((System.currentTimeMillis() - startTime) / 1000);
}
}
和传统的生产者消费者不同，这里实际上只有消费者，因为产生原始数据几乎不耗时，最容易想到的办法就是定义个共享的index标志，依次互斥的进行+1操作，因此这里的index就是一个共享的变量，需要进行同步。直接使用jdk中提供的AtomicInteger，代码如下：
package tool;
import java.util.List;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.atomic.AtomicInteger;
public class MutiThread
{
private static AtomicInteger lock = new AtomicInteger(0);
public static void main(String[] args)
{
MutiThread tool = new MutiThread();
String userIdPath = "D:\\shell\\store_bak\\tool\\userid.txt";
List<String> userIds = Util.readUserId(userIdPath);
List<String> cdrItems = Util.readCdrItem();
tool.work2(lock, userIds, cdrItems);
}
public void work2(AtomicInteger lock, List<String> userIds,
List<String> cdrItems)
{
final long startTime = System.currentTimeMillis();
CyclicBarrier cb = new CyclicBarrier(5, new Runnable()
{
@Override
public void run()
{
System.out.println((System.currentTimeMillis() - startTime) / 1000);
}
});
for (int i = 0; i < 5; i++)
{
new Thread(new Worker(userIds, cdrItems, lock, cb)).start();
}
}
class Worker implements Runnable
{
private List<String> userIds;
private List<String> cdrItems;
private AtomicInteger lock;
private CyclicBarrier cb;
public Worker(List<String> userIds, List<String> cdrItems,
AtomicInteger lock, CyclicBarrier cb)
{
this.userIds = userIds;
this.cdrItems = cdrItems;
this.lock = lock;
this.cb = cb;
}
@Override
public void run()
{
while (true)
{
int index = lock.getAndIncrement();
if (index >= userIds.size())
break;
String id = userIds.get(index);
process1(id, cdrItems);
}
try
{
cb.await();
} catch (InterruptedException e)
{
e.printStackTrace();
} catch (BrokenBarrierException e)
{
e.printStackTrace();
}
}
}
private void process1(String id, List<String> cdrItems)
{
String[] uninKeys = id.split("\\s+");
int count = 0;
for (String cdr : cdrItems)
{
if (cdr.contains("|" + uninKeys[0] + "|")
&& cdr.contains("|" + uninKeys[1] + "|"))
{
count++;
}
}
}
}
针对这个场景，考虑下concurrenthashmap的实现，可以将资源进行分段处理，可以巧妙的避开多线程的资源征用，因此可以将list分成不同的段，交给不同的线程去处理，代码如下：
package tool;
import java.util.List;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.atomic.AtomicInteger;
public class MutiSegmentMutiThread
{
private static AtomicInteger lock = new AtomicInteger(0);
private static int ThreadNum = 10;
public static void main(String[] args)
{
MutiSegmentMutiThread tool = new MutiSegmentMutiThread();
String userIdPath = "D:\\shell\\store_bak\\tool\\userid.txt";
List<String> userIds = Util.readUserId(userIdPath);
List<String> cdrItems = Util.readCdrItem();
tool.work2(lock, userIds, cdrItems);
}
public void work2(AtomicInteger lock, List<String> userIds,
List<String> cdrItems)
{
final long startTime = System.currentTimeMillis();
CyclicBarrier cb = new CyclicBarrier(ThreadNum, new Runnable()
{
@Override
public void run()
{
System.out.println((System.currentTimeMillis() - startTime) / 1000);
}
});
int segmentSize = userIds.size() / ThreadNum;
int start = 0;
int end = 0;
for (int i = 0; i < ThreadNum; i++)
{
start = i * segmentSize;
if (i == ThreadNum - 1)
{
end = userIds.size();
} else
{
end = (i + 1) * segmentSize;
}
new Thread(new Worker(userIds, cdrItems, cb, start, end)).start();
}
}
class Worker implements Runnable
{
private List<String> userIds;
private List<String> cdrItems;
private CyclicBarrier cb;
private int start;
private int end;
public Worker(List<String> userIds, List<String> cdrItems,
CyclicBarrier cb, int start, int end)
{
this.userIds = userIds;
this.cdrItems = cdrItems;
this.cb = cb;
this.start = start;
this.end = end;
}
@Override
public void run()
{
for (int i = start; i < end; i++)
{
String id = userIds.get(i);
process1(id, cdrItems);
}
try
{
cb.await();
} catch (InterruptedException e)
{
e.printStackTrace();
} catch (BrokenBarrierException e)
{
e.printStackTrace();
}
}
}
private void process1(String id, List<String> cdrItems)
{
String[] uninKeys = id.split("\\s+");
int count = 0;
for (String cdr : cdrItems)
{
if (cdr.contains("|" + uninKeys[0] + "|")
&& cdr.contains("|" + uninKeys[1] + "|"))
{
count++;
}
}
}
}
原文地址：http://blog.csdn.net/tangyongzhe/article/details/43424707
