Python多线程 VS JAVA多线程

yanfei

      最近在学习Python，在网上遇到一哥们问一个关于用Python多线程的问题。具体就是读取一个文件中的IP信息，然利用百度地图open api获取IP所在地的经纬度，最后再将取回的结果分类为“请求成功”和“请求失败”分别存入两个文件中。由于自己也在学，所以也就拿来做了做，并分别用Python3和Java实现。
    稍稍分析了一下需求能看的出来里面性能瓶颈有两个：
    1、一个是查询百度地图OPEN API；
    2、写磁盘文件；
    于是想到把这两部分分开处理，自然的想到了生产-消费者模式来实现。
    对于IP文件信息格式如下（我的测试文件有41255行数据）：
   
　　61.0.0.0__61.255.255.255__亚洲____
　　61.157.0.0__61.157.255.255__四川____
　　61.139.0.0__61.139.127.255__四川____
　　61.130.93.0__61.130.254.255__浙江____
    下面是Python3实现代码：

#coding:utf-8
import urllib.request
import json
import threading
import queue
FILE_ENCODING = "utf-8"
successList = []
failList = []
successPath = r"success.txt"
failPath = r"failed.txt"
ipPath = "ip.txt"
queryCount = 5
threadCount = 20
lock = threading.RLock()
separator = "|"
saveCount = 100
successIndex = 0
failIndex = 0
theCount = 0

queryQueue = queue.Queue()
processedQueue = queue.Queue()
class IPStruct(object):
def __init__(self):
self.startIP = None
self.endIP = None
self.address = ""
self.position = ""
self.status = ""

class QueryIPLocation(object):
_retry = 5
def query(self, objList):
for obj in objList:
url = self.get_url(obj)#.encode('utf-8')
#            print("url:%s" % url)
for j in range(0, self._retry + 1):
try:
result = urllib.request.urlopen(url,timeout=5)
except Exception:
if j == self._retry:
obj.status = "Failed"
break
continue
if result.getcode() != 200:
if j == self._retry:
obj.status = "Failed"
else:
jsonRet = json.loads(result.read().decode())
if jsonRet['status'] == 0:
obj.status = "SUCCESS"
self.proc_ok(jsonRet, obj)
break
elif jsonRet['status'] == 1:
obj.status = "SERVER_ERROR"
break
elif jsonRet['status'] == 2:
obj.status = "INVALID_PARAMETER"
break
elif jsonRet["status"] == 5:
obj.status = "INVALID_AK"
break
elif jsonRet["status"] == 101:
obj.status = "DISABLE_SERVICE"
break
elif jsonRet['status'] == 345:
if j == self._retry:
obj.status = "MINUTE_QUERY_LIMIT"
elif jsonRet["status"] == 355:
obj.status = "DAY_QUERY_LIMIT"
break
def get_url(self, obj):
base = "http://api.map.baidu.com/location/ip?ak=7E814968d7b3ee0440678cb17cb4aa29&coor=bd09ll"
return base + "&ip=" + str(obj.startIP)
def proc_ok(self, result, obj):
point = result["content"]["point"]
address = result["content"]["address"]
obj.address = address
obj.position = str(point['x']) + "," + str(point['y'])
class ThreadUrl(threading.Thread):
"""Threaded Url Grab"""
def __init__(self,queryQueue,processedQueue):
threading.Thread.__init__(self)
self.queryQueue = queryQueue
self.processedQueue = processedQueue
def run(self):
while True:
queryList = self.queryQueue.get()
app = QueryIPLocation()
try:
app.query(queryList)
for item in queryList:
self.processedQueue.put(item)
except Exception:
print("there has error..........")
raise
finally:
self.queryQueue.task_done()
class ThreadFile(threading.Thread):
"""Threaded save File"""
def __init__(self,queue):
threading.Thread.__init__(self)
self.queue = queue
def run(self):
global successList
global failList
global successIndex
global failIndex
global theCount
while True:
item = self.queue.get()
if item.status == "SUCCESS":
lock.acquire()
try:
theCount += 1
print("theCount:",theCount)
successList.append(item)
successIndex += 1
if successIndex == saveCount:
save_result(successPath,resultSequence=successList)
successIndex = 0
del successList[:]
except Exception:
raise
finally:
lock.release()
else:
lock.acquire()
try:
theCount += 1
print("theCount:",theCount)
failList.append(item)
failIndex += 1
if failIndex == saveCount:
save_result(failPath,resultSequence=failList)
failIndex = 0
del failList[:]
except Exception:
raise
finally:
lock.release()
self.queue.task_done()
def save_result(filePath, resultSequence=[]):
"""save the success result to successPath"""
line = ""
file = open(filePath, "a")
if not file:
print("error open file %s to write" % filePath)
for item in resultSequence:
if item.status == "SUCCESS":
line += item.startIP + separator + item.endIP + separator + item.address + separator + item.position + "\n"
else:
line += item.startIP + separator + item.endIP + separator + item.address + separator + item.status + "\n"
file.write(line)
file.close()
def get_ip_iter(file):
ipList = []
for line in file:
addr = line.split("__")
count = len(addr)
if count <= 2 or "." not in addr[0] or "." not in addr[1]:
continue
obj = IPStruct()
obj.startIP = addr[0]
obj.endIP = addr[1]
obj.address = addr[2]
ipList.append(obj)
if len(ipList) == queryCount:
yield ipList
del ipList[:]
else:
if len(ipList) > 0:
yield ipList
def main():
print("start read file")
ipFile = open(ipPath,"r",encoding="utf-8")
if not ipFile:
print("errror open file %s" % ipPath)
return
iterIpFile = get_ip_iter(ipFile)
for i in range(threadCount):
t = ThreadUrl(queryQueue,processedQueue)
t.setDaemon(True)
t.start()
for queryList in iterIpFile:
queryQueue.put(list(queryList))
for i in range(2):
dt = ThreadFile(processedQueue)
dt.setDaemon(True)
dt.start()
queryQueue.join()
processedQueue.join()
save_result(successPath,resultSequence=successList)
save_result(failPath,resultSequence=failList)
    使用 time python3 ***.py 执行：

real    15m46.962s
user    0m38.042s
sys     1m0.162s
    可以看到运行时间15分46秒
 
    由于自己以前一直学的是Java，随即就对用Python和用Java的多线程来实现谁的执行速度会更高一些产生一些性趣，索性就会就Java也实现了同样的需求，并使用了同样的线程数。
    以下是JAVA实现的代码：
    

package test;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import net.sf.json.JSONObject;
import util.HttpUtil;
import util.JSONUtil;
public class IPQueryTest {

public static void main(String[] args) throws InterruptedException {
long start = System.currentTimeMillis();
BlockingQueue<IPStruct> fileQueue = new LinkedBlockingQueue<IPStruct>();
BlockingQueue<IPStruct> ipQueue = new LinkedBlockingQueue<IPStruct>();
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new ThreadReadFile(fileQueue));
for(int i=0;i<20;i++){
exec.execute(new ThreadQueryIPLocation(fileQueue,ipQueue));
}
for(int i=0;i<2 ;i++){
exec.execute(new ThreadWriteFile(ipQueue));
}
while(true){
Thread.sleep(10);
if(fileQueue.size() == 0 && ipQueue.size() == 0){
ThreadWriteFile.saveResult(ThreadWriteFile.SUCC_PATH, ThreadWriteFile.succIPStructs);
ThreadWriteFile.saveResult(ThreadWriteFile.FAIL_PATH, ThreadWriteFile.failIPStructs);
long stend = System.currentTimeMillis();
System.out.println("total cost:" + (stend - start));
break;
}
/*else{
System.out.println("fileQueue size:" + fileQueue.size() + ",ipQueue size:" + ipQueue.size());
}*/
}
}
}
class IPStruct {
private String startIP;
private String endIP;
private String address;
private String position;
private String status;
public String getStartIP() {
return startIP;
}
public void setStartIP(String startIP) {
this.startIP = startIP;
}
public String getEndIP() {
return endIP;
}
public void setEndIP(String endIP) {
this.endIP = endIP;
}
public String getAddress() {
return address;
}
public void setAddress(String address) {
this.address = address;
}
public String getPosition() {
return position;
}
public void setPosition(String position) {
this.position = position;
}
public String getStatus() {
return status;
}
public void setStatus(String status) {
this.status = status;
}
}
class ThreadReadFile implements Runnable{
private final BlockingQueue<IPStruct> queue;
private static final String FILE_PATH = "D:\\ip.txt";
ThreadReadFile(BlockingQueue<IPStruct> q){
this.queue = q;
}
@Override
public void run() {
try{
produceIPS(queue);
}catch(InterruptedException ex){
ex.printStackTrace();
}
}
public void produceIPS(BlockingQueue<IPStruct> queue)throws InterruptedException{
File file = null;
BufferedReader reader = null;
try{
file = new File(FILE_PATH);
reader = new BufferedReader(new FileReader(file));
String line = null;
IPStruct ipStruct = null;
while((line = reader.readLine()) != null){
ipStruct = formatIPStruct(line);
if(null != ipStruct){
queue.add(ipStruct);
}
}
}catch(IOException e){
e.printStackTrace();
}finally{
if(null != reader){
try {
reader.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
public IPStruct formatIPStruct(String line){
String[] addr = line.split("__");
if(addr.length <= 2 || addr[0].indexOf(".")== -1 || addr[1].indexOf(".") == -1){
return null;
}
IPStruct ipStruct = new IPStruct();
ipStruct.setStartIP(addr[0]);
ipStruct.setEndIP(addr[1]);
ipStruct.setAddress(addr[2]);
return ipStruct;
}
}
class ThreadQueryIPLocation implements Runnable{
private final BlockingQueue<IPStruct> inputQueue;
private final BlockingQueue<IPStruct> outputQueue;
ThreadQueryIPLocation(BlockingQueue<IPStruct> inputQueue,BlockingQueue<IPStruct> outputQueue){
this.inputQueue = inputQueue;
this.outputQueue = outputQueue;
}
@Override
public void run() {
try{
while(true){
IPStruct ipStruct = inputQueue.take();
queryIPLocation(ipStruct);
outputQueue.add(ipStruct);
}
}
catch(InterruptedException ex){
ex.printStackTrace();
}
}
private static final int RETRY = 5;
private static final String BASE_URL = "http://api.map.baidu.com/location/ip?ak=7E814968d7b3ee0440678cb17cb4aa29&coor=bd09ll";
public void queryIPLocation(IPStruct obj) {
String url = null;
String result = null;
url = BASE_URL + "&ip=" + obj.getStartIP();
//System.out.println("url-->" + url);
for (int j = 0; j <= RETRY; j++) {
try {
result = HttpUtil.getJsonStringFromUrl(url);
} catch (Exception e) {
if (j == RETRY) {
obj.setStatus("Failed");
break;
}
continue;
}
if (null != result) {
JSONObject jsonObject = JSONUtil.toJSONObject(result);
Integer status = (Integer) jsonObject.get("status");
if (status == 0) {
obj.setStatus("SUCCESS");
procOK(jsonObject, obj);
break;
} else if (status == 1) {
obj.setStatus("SERVER_ERROR");
break;
} else if (status == 5) {
obj.setStatus("INVALID_AK");
break;
} else if (status == 101) {
obj.setStatus("DISABLE_SERVICE");
break;
} else if (status == 345) {
if (j == RETRY) {
obj.setStatus("MINUTE_QUERY_LIMIT");
}
} else if (status == 355) {
obj.setStatus("DAY_QUERY_LIMIT");
break;
}
}
}
}
public void procOK(JSONObject result, IPStruct obj) {
JSONObject contentJSON = JSONUtil.toJSONObject(result.get("content"));
JSONObject pointJSON = JSONUtil.toJSONObject(contentJSON.get("point").toString());
obj.setAddress((String) contentJSON.get("address"));
obj.setPosition(pointJSON.get("x") + "," + pointJSON.get("y"));
}
}
class ThreadWriteFile implements Runnable{
public static final List<IPStruct> succIPStructs = new ArrayList<IPStruct>();
public static final List<IPStruct> failIPStructs = new ArrayList<IPStruct>();
public static final String SEPARATOR = "|";
public static final int FILE_SAVE_SIZE = 100;
public static final String SUCC_PATH = "D:\\success.txt";
public static final String FAIL_PATH = "D:\\fail.txt";
public static AtomicInteger counter = new AtomicInteger(0);

private final BlockingQueue<IPStruct> queue;
ThreadWriteFile(BlockingQueue<IPStruct> queue){
this.queue = queue;
}
@Override
public void run() {
try{
while(true){
IPStruct ipStruct = queue.take();
count();
if(ipStruct.getStatus() == "SUCCESS"){
saveSuccIPStruct(ipStruct);
}
else{
saveFailIPStruct(ipStruct);
}
}
}
catch(InterruptedException ex){
ex.printStackTrace();
}
}
private synchronized void count(){
counter.addAndGet(1);
System.out.println("count:" + counter.get());
}
public static void writeResult(String lines,String path){
try {
if(null == lines || lines.trim().equals("")){
return;
}
FileWriter writer = new FileWriter(path,true);
writer.write(lines);
writer.close();
} catch (IOException e) {
e.printStackTrace();
}
}
public synchronized void saveSuccIPStruct(IPStruct ipStruct){
succIPStructs.add(ipStruct);
if(succIPStructs.size() >= FILE_SAVE_SIZE){
StringBuffer lines = new StringBuffer();
for(IPStruct succIPStruct : succIPStructs){
lines.append(succIPStruct.getStartIP()).append(SEPARATOR);
lines.append(succIPStruct.getEndIP()).append(SEPARATOR);
lines.append(succIPStruct.getAddress()).append(SEPARATOR);
lines.append(succIPStruct.getPosition()).append("\n");
}
writeResult(lines.toString(),SUCC_PATH);
succIPStructs.clear();
}
}
public synchronized void saveFailIPStruct(IPStruct ipStruct){
failIPStructs.add(ipStruct);
if(failIPStructs.size() >= FILE_SAVE_SIZE){
StringBuffer lines = new StringBuffer();
for(IPStruct failIPStruct : failIPStructs){
lines.append(failIPStruct.getStartIP()).append(SEPARATOR);
lines.append(failIPStruct.getEndIP()).append(SEPARATOR);
lines.append(failIPStruct.getAddress()).append(SEPARATOR);
lines.append(failIPStruct.getStatus()).append("\n");
}
writeResult(lines.toString(),FAIL_PATH);
failIPStructs.clear();
}
}
public static void saveResult(String path,List<IPStruct> ipStructs){
StringBuffer lines = new StringBuffer();
for(IPStruct ipStruct : ipStructs){
lines.append(ipStruct.getStartIP()).append(SEPARATOR);
lines.append(ipStruct.getEndIP()).append(SEPARATOR);
lines.append(ipStruct.getAddress()).append(SEPARATOR);
if(ipStruct.getStatus() == "SUCCESS"){
lines.append(ipStruct.getPosition()).append("\n");
}
else{
lines.append(ipStruct.getStatus()).append("\n");
}
}
writeResult(lines.toString(),path);
}
}

    JAVA代码是在windows下eclipse里执行：
    

total cost:1132948
    可以看出JAVA代码的运行时间大约在11分钟。单单从运行时间上看java的实现会快差不多4分钟，大师们也都说过Python很关注程序员的生产率而不是绝对的性能。当然很大的可能性是我代码本身问题，而造成这4分钟的时间差。
    写这些也并不想真的比较python和java那个好，只是想这样可以有对比的学习，做以记录。