java实现服务端守护进程来监听客户端通过上传json文件写数据到hbase中
1、项目介绍：
由于大数据部门涉及到其他部门将数据传到数据中心，大部分公司采用的方式是用json文件的方式传输，因此就需要编写服务端和客户端的小程序了。而我主要实现服务端的代码，也有相应的客户端的测试代码。这里须有一个需要提到的是，我在实现接收json文件的同时，而且还需将数据写到hbase中。写入到hbase当中采用的是批量插入的方式，即一次插入多条记录。
好了，有了前面的说明，下面来简单的说一下我实现的服务端的小程序把。
2、为了实现服务端能够监听客户端的行为，因此我在服务端采用多线程的技术来实现，并用socket的方式来实现网络通信。具体实现如下：
服务端的主程序：
package com.yiban.datacenter.finalversion;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
public class HbaseServer {
public static void main(String[] args) {
// TODO Auto-generated method stub
backprocess();
}
public static void backprocess(){
try {
ServerSocket ss=new ServerSocket(11111);
while(true){
Socket s=ss.accept();
Thread deal=new Thread(new DealUserThread(s));
deal.setDaemon(true); //这里设置对应线程是后台线程
deal.start();
}
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
处理数据的线程类：
在这里我实现了接收数据，并将数据写入hbase中。
在实现这些大的目标的同时，也将客户端的请求通过日志文件的形式存到服务端的本地磁盘上，供后续查看。
package com.yiban.datacenter.finalversion;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.Socket;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
public class DealUserThread implements Runnable {
private String testconnect = "username=chenpiao,password=123;username=liujiyu,password=123"; // 这个可以用来验证用户名和密码
private static Configuration conf = HBaseConfiguration.create();
private static Connection connection = null;
private String logFile=null;
// 配置hbase的信息
static {
try {
conf.set(HConstants.ZOOKEEPER_QUORUM, "192.168.27.233");
conf.setInt(HConstants.ZOOKEEPER_CLIENT_PORT, 2181);
connection = ConnectionFactory.createConnection(conf);
} catch (Exception e) {
// TODO: handle exception
e.printStackTrace();
}
}
private Socket s;
public DealUserThread(Socket s) {
this.s = s;
}
private String userTableName = "nihao";
private String columnFamilyName = null;
private String rowKey = null;
private BufferedReader serverread = null;
private BufferedWriter serverwrite = null;
@Override
public void run() {
// TODO Auto-generated method stub
try {
// 将通道内的字节流转换成字符流，并用bufferedreader进行封装，InputStreamReader是将字节流转换成字符流
serverread = new BufferedReader(new InputStreamReader(
s.getInputStream()));
// 询问客户端连接是否准备好，接受客户端的连接请求
String line = serverread.readLine(); // 阻塞
//System.out.println(line);// 输出客户端的连接请求
//为日志文件命名,并创建文件
SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
logFile="/var/log/datacenter/"+"user.log";
//System.out.println(logFile);
File destFile = new File(logFile);
if (!destFile.exists()) {
destFile.createNewFile();
}
writeByFileWrite(logFile, line+"
"+sdf.format(System.currentTimeMillis()));
// 将通道内的字符写入到对应的文件中，利用bufferedwrite进行封装,FileWriter是将字符流写入到文件中
serverwrite = new BufferedWriter(new OutputStreamWriter(
s.getOutputStream()));
String[] strArray = testconnect.split("\;");
boolean flag = false;
for (String str : strArray) {
if (str.equals(line)) {
/*
* serverwrite.write("连接成功，你可以发送数据了，发送数据前，请先发送你要用的数据库表名！");
* serverwrite.newLine(); serverwrite.flush();
*/
printInfomationForClient("connection successful ,now you can send data,befor send data ,you must send tablename!");
flag = true;
break;
}
}
if (!flag) {
/*
* serverwrite.write("密码或者用户名错误，连接失败！"); serverwrite.newLine();
* serverwrite.flush();
*/
printInfomationForClient("username or password is error! connection failed!");
s.close();
}
// 准备接收表名
userTableName = serverread.readLine();
//System.out.println("tablename:" + userTableName);// 输出客户端的连接请求的表名
writeByFileWrite(logFile, "tablename="+userTableName);//将内容写到日志文件中
// 告诉客户端，我接受成功
if (TableIsExist(userTableName)) {
printInfomationForClient("received tablename successful!");
} else {
printInfomationForClient("tablename Is not exist ");
s.close();
}
line = "[";
StringBuffer temp = new StringBuffer(line);
while ((line = serverread.readLine()) != null) {
temp.append(line);
}
temp.append("]");
//System.out.println(temp.toString());
// 对接收到的数据进行异常处理，如上传的数据格式不正确等等。
try {
// 对json文件进行解析
JSONArray jsonArray = JSONArray.fromObject(temp.toString());
// JSONObject jsonobject=JSONObject.fromObject(temp.toString());
// 解析之后进行输出
//PrintJsonArray(jsonArray);
//获取所有的表名
//getAllTables(conf);
// 将接收到的数据写入hbase中的表中
insertData(jsonArray, userTableName);
//System.out.println("存入数据成功！");
} catch (Exception e) {
printErrorForClient(e);
}
// 给出一个反馈，提示数据上传成功
// 封装通道内的输出流，方便对他进行写字符数据
// BufferedWriter bwserver = new BufferedWriter(new
// OutputStreamWriter(s.getOutputStream()));
/*
* serverwrite.write("文件上传成功！"); // bwserver.newLine();
* serverwrite.flush(); serverwrite.close();
*/
printInfomationForClient("upload data successful!");
// 释放资源
s.close();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
try {
printErrorForClient(e);
writeByFileWrite(logFile,e.getMessage()+e.toString());
} catch (IOException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
}
}
private void printInfomationForClient(String s) throws IOException {
try {
serverwrite.write(s);
writeByFileWrite(logFile, s);//将内容写到日志文件中
serverwrite.newLine();
serverwrite.flush();
} catch (Exception e) {
// TODO: handle exception
e.printStackTrace();
writeByFileWrite(logFile, e.getMessage()+e.toString());//将内容写到日志文件中
}
}
private void printErrorForClient(Exception e) throws IOException {
try {
serverwrite.write("found a error:" + e.getMessage() + e.toString());
writeByFileWrite(logFile,"found a error:" + e.getMessage() + e.toString());
serverwrite.newLine();
serverwrite.flush();
s.close();
} catch (IOException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
writeByFileWrite(logFile,e1.getMessage()+e1.toString());
}
}
Map<String, String> colvalue = new TreeMap<String, String>();
private void insertData(JSONArray jsonArray, String userTableName)
throws IOException {
// connect the table
Table table = null;
try {
table = connection.getTable(TableName.valueOf(userTableName));
} catch (IOException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
printErrorForClient(e1);
}
colvalue.clear();
for (int i = 0; i < jsonArray.size(); i++) {
JSONObject obj = jsonArray.getJSONObject(i);
Set<String> keysets = obj.keySet();
for (String key : keysets) {
switch (key) {
case "category":
columnFamilyName = obj.getString(key);
break;
case "version":
colvalue.put("version", obj.getString("version"));
break;
case "DocumentType":
colvalue.put("DocumentType", obj.getString("DocumentType"));
break;
case "articles":
JSONArray articlesjars = obj.getJSONArray("articles");
dealjsonArray(table,articlesjars);
break;
default:
printErrorForClient(new Exception("send datatype is error!"));
}
}
}
}
private void insertColDataToHbase(Table table) throws IOException {
// 判断是否包含对应的列族，若不包含则添加
HTableDescriptor desc = new HTableDescriptor(table.getName());
Collection<HColumnDescriptor> familys=desc.getFamilies();
if ( familys.contains(new HColumnDescriptor(columnFamilyName))
&& columnFamilyName != null) {
addColFamily(table, desc, columnFamilyName);
}
// insert data
List<Put> putlist = new ArrayList<Put>();
if (!colvalue.isEmpty() && rowKey != null && columnFamilyName != null) {
Put put = new Put(Bytes.toBytes(rowKey));// 指定行,也就是键值
// 下面就是循环存储列
for (Entry<String, String> col : colvalue.entrySet()) {
put.add(Bytes.toBytes(columnFamilyName),
Bytes.toBytes(col.getKey()),
Bytes.toBytes(col.getValue()));
putlist.add(put);
}
} else {
printErrorForClient(new Exception("send datatype is error!"));
}
try {
table.put(putlist);
table.close();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
printErrorForClient(e);
}
}
private void dealjsonArray(Table table,JSONArray articlesjars) throws IOException {
// TODO Auto-generated method stub
if(articlesjars.isEmpty()){
System.out.println("articlesjars is empty");
printErrorForClient(new Exception("send datatype is error!"));
return;
}
for(int i=0;i<articlesjars.size();i++){
JSONObject obj=articlesjars.getJSONObject(i);
Set<String>keysets=obj.keySet();
for(String key:keysets){
switch(key){
case "content":
colvalue.put("content", obj.getString("content"));
break;
case "picture_url":
colvalue.put("picture_url", obj.getString("picture_url"));
break;
case "time":
colvalue.put("time", obj.getString("time"));
break;
case "author":
colvalue.put("author", obj.getString("author"));
break;
case "url":
rowKey=obj.getString("url");
break;
case "title":
colvalue.put("title", obj.getString("title"));
break;
default:
printErrorForClient(new Exception("send datatype is error!"));
}
}
try {
insertColDataToHbase(table);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
printErrorForClient(e);
}
}
}
private void addColFamily(Table table, HTableDescriptor desc,
String colFamily) throws IOException {
Admin ad = connection.getAdmin();
HColumnDescriptor family = new HColumnDescriptor(
Bytes.toBytes(colFamily));// 列簇
desc.addFamily(family);
ad.addColumn(table.getName(), family);
ad.close();
}
private boolean TableIsExist(String userTableName2) throws IOException {
boolean flag = false;
try {
// Connection connection = ConnectionFactory.createConnection(conf);
Admin ad = connection.getAdmin();
if (ad.tableExists(TableName.valueOf(userTableName2))) {
flag = true;
//System.out.println("表存在");
} else {
//System.out.println("表不存在");
//printErrorForClient(new Exception("表不存在"));
}
} catch (Exception e) {
// TODO: handle exception
e.printStackTrace();
printErrorForClient(e);
}
return flag;
// TODO Auto-generated method stub
}
private void PrintJsonArray(JSONArray jsonArray) {
int size = jsonArray.size();
System.out.println("Size: " + size);
for (int i = 0; i < size; i++) {
JSONObject jsonObject = jsonArray.getJSONObject(i);
Set<String> keysets = jsonObject.keySet();
for (String keyset : keysets) {
System.out.println(keyset);
}
}
}
private void PrintJsonArray(JSONObject jsonobject, String... keys) {
int size = jsonobject.size();
System.out.println("Size: " + size);
for (int i = 0; i < size; i++) {
for (String key : keys) {
System.out.println(key + ":" + jsonobject.get(key));
}
// System.out.println("[" + i + "]));
// System.out.println("[" + i + "]name=" + jsonObject.get("name"));
// System.out.println("[" + i + "]role=" + jsonObject.get("role"));
}
}
//写日志文件
public static void writeByFileWrite(String _sDestFile, String _sContent)
throws IOException {
FileWriter fw = null;
try {
fw = new FileWriter(_sDestFile,true);
fw.write(_sContent);
fw.write('
');
} catch (Exception ex) {
ex.printStackTrace();
} finally {
if (fw != null) {
fw.close();
fw = null;
}
}
}
// create table
private void createTable(Configuration conf) {
// HBaseAdmin ha=new HBaseAdmin(conf);
try {
// Connection connection = ConnectionFactory.createConnection(conf);
Table table = connection.getTable(TableName.valueOf(userTableName));
Admin ad = connection.getAdmin();
// TableName name= TableName.valueOf(Bytes.toBytes(tablename));//表名
HTableDescriptor desc = new HTableDescriptor(table.getName());
HColumnDescriptor family = new HColumnDescriptor(
Bytes.toBytes(columnFamilyName));// 列簇
desc.addFamily(family);
ad.createTable(desc);
ad.close();
} catch (Exception e) {
// TODO: handle exception
e.printStackTrace();
}
}
// Hbase获取所有的表信息
public static List getAllTables(Configuration conf)
throws MasterNotRunningException, ZooKeeperConnectionException,
IOException {
HBaseAdmin ad = new HBaseAdmin(conf);
List<String> tables = null;
if (ad != null) {
try {
HTableDescriptor[] allTable = ad.listTables();
if (allTable.length > 0)
tables = new ArrayList<String>();
for (HTableDescriptor hTableDescriptor : allTable) {
tables.add(hTableDescriptor.getNameAsString());
System.out.println(hTableDescriptor.getNameAsString());
}
} catch (IOException e) {
e.printStackTrace();
}
}
return tables;
}
}
实现了这些之后，通过eclipse将其导出程可运行的jar包，并将jar包放到服务器上进行部署，部署的方式很简单，但是也要注意一下：
java -jar myserver.jar com.yiban.datacenter.finalversion.HbaseServer &
最后一个&表示后台守护启动该进程。