p49023857
/
r-memcached
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'Memcached_Client'

This commit is contained in:
lyr90329 2014-08-13 17:57:36 +08:00
parent 43b0aa5847 4f1b963fcf
commit 09199591ca
7 changed files with 321 additions and 367 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
MemcachedClient/src/common/EMSGID.java
View File

@ -16,5 +16,8 @@ public enum EMSGID
nm_read,
nm_write_1,
nm_write_1_res,
nm_write_2
nm_write_2,
nr_stats,
nr_stats_res
}

2
MemcachedClient/src/common/RegisterHandler.java
View File

@ -4,6 +4,7 @@ import java.lang.reflect.InvocationTargetException;
import messageBody.memcachedmsg.nm_Connected;
import messageBody.requestMsg.nr_Connected_mem_back;
import messageBody.requestMsg.nr_Read_res;
import messageBody.requestMsg.nr_Stats_res;
import messageBody.requestMsg.nr_write_res;
import com.google.protobuf.GeneratedMessage;
@ -16,6 +17,7 @@ public class RegisterHandler
initHandler(EMSGID.nm_connected.ordinal(), nm_Connected.class);
initHandler(EMSGID.nr_connected_mem_back.ordinal(), nr_Connected_mem_back.class);
initHandler(EMSGID.nr_read_res.ordinal(), nr_Read_res.class);
initHandler(EMSGID.nr_stats_res.ordinal(), nr_Stats_res.class);
initHandler(EMSGID.nr_write_res.ordinal(), nr_write_res.class);
}

36
MemcachedClient/src/server/MDecoder.java
View File

@ -1,35 +1,33 @@
package server;
package com.myself.server;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.handler.codec.frame.FrameDecoder;
import server.NetMsg;
import common.EMSGID;
public class MDecoder extends FrameDecoder
{
@Override
protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) throws Exception {
if (buffer.readableBytes() < 4)
{
return null;//(1)
}
int dataLength = buffer.getInt(buffer.readerIndex());
if (buffer.readableBytes() < dataLength + 4)
{
return null;//(2)
}
if (buffer.readableBytes() < 8) {
return null;// (1)
}
int dataLength = buffer.getInt(buffer.readerIndex());
if (buffer.readableBytes() < dataLength + 4) {
return null;// (2)
}
buffer.skipBytes(4);//(3)
int id = buffer.readInt();
byte[] decoded = new byte[dataLength-4];
buffer.skipBytes(4);// (3)
int id = buffer.readInt();
int nodeRoute = buffer.readInt();
byte[] decoded = new byte[dataLength - 8];
buffer.readBytes(decoded);
NetMsg msg = new NetMsg(decoded, id);//(4)
msg.setMsgID(EMSGID.values()[id]);
return msg;
buffer.readBytes(decoded);
NetMsg msg = new NetMsg(decoded, id);// (4)
msg.setMsgID(EMSGID.values()[id]);
msg.setNodeRoute(nodeRoute);
return msg;
}
}

9
MemcachedClient/src/server/MEncoder.java
View File

@ -1,4 +1,4 @@
package server;
package com.myself.server;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
@ -6,23 +6,22 @@ import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.handler.codec.oneone.OneToOneEncoder;
import server.NetMsg;
public class MEncoder extends OneToOneEncoder
{
@Override
protected Object encode(ChannelHandlerContext ctx, Channel channel, Object msg) throws Exception
{
if (!(msg instanceof NetMsg))
if (!(msg instanceof NetMsg))
{
return msg;//(1)
}
NetMsg res = (NetMsg)msg;
byte[] data = res.getBytes();
int dataLength = data.length+4;
int dataLength = data.length+8;
ChannelBuffer buf = ChannelBuffers.dynamicBuffer();//(2)
buf.writeInt(dataLength);
buf.writeInt(res.msgID.ordinal());
buf.writeInt(res.getNodeRoute());
buf.writeBytes(data);
return buf;//(3)
}

15
MemcachedClient/src/server/NetMsg.java
View File

@ -1,7 +1,6 @@
package server;
package com.myself.server;
import com.google.protobuf.GeneratedMessage;
import com.google.protobuf.MessageLite;
import server.NetMsg;
import common.EMSGID;
import common.MessageManager;
@ -10,11 +9,14 @@ public class NetMsg
{
EMSGID msgID;
MessageLite messageLite;
int nodeRoute;
private NetMsg(){};
public static NetMsg newMessage()
{
return new NetMsg();
NetMsg msg = new NetMsg();
msg.setNodeRoute(0);
return msg;
}
NetMsg(byte[] decoded, int id) throws Exception
@ -49,5 +51,10 @@ public class NetMsg
{
this.messageLite = builder.build();
}
public int getNodeRoute() {
return nodeRoute;
}
public void setNodeRoute(int nodeRoute) {
this.nodeRoute = nodeRoute;
}
}

243
MemcachedClient/src/server/WebServerMain.java
View File

@ -1,10 +1,8 @@
package server;
package com.myself.server;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.HashMap;
import java.util.Properties;
import java.util.Random;
import javax.xml.parsers.DocumentBuilder;
@ -12,51 +10,44 @@ import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;
import org.apache.log4j.PropertyConfigurator;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.xml.sax.SAXException;
import r_memcached.MemcachedMgr;
import server.ClientConfig;
import server.Server;
import com.myself.memcached.MemcachedMgr;
import common.RegisterHandler;
public class WebServerMain
{
public class WebServerMain {
static HashMap<Integer, ClientConfig> m_mapMemcachedClient;
public static boolean initConfig()
{
static int threadCount = 0;
public static boolean initConfig() {
m_mapMemcachedClient = new HashMap<Integer, ClientConfig>();
File f = new File(System.getProperty("user.dir"));
String path = f.getPath() + File.separator + "bin" + File.separator;
readClientsXML(path+"client.xml");
try {
Properties properties = new Properties();
properties.load(new FileInputStream(path+"config.properties"));
MemcachedMgr.nCopyNode = Integer.parseInt(properties.getProperty("replicasNum"));
} catch (FileNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
readClientsXML(path + "client.xml");
return true;
}
private static class bench extends Thread
{
private static class bench extends Thread {
private int runs;
@SuppressWarnings("unused")
private int threadNum;
private String object;
private String[] keys;
@SuppressWarnings("unused")
private int size;
private int nums;
private double rate;
public bench(int runs,int nums, int threadNum, String object, String[] keys, double rate)
{
public bench(int runs, int nums, int threadNum, String object,
String[] keys, double rate) {
this.runs = runs;
this.threadNum = threadNum;
this.object = object;
@ -66,13 +57,10 @@ public class WebServerMain
this.rate = rate;
}
public void run()
{
try
{
public void run() {
try {
Thread.sleep(10);
} catch (InterruptedException e)
{
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
@ -81,28 +69,22 @@ public class WebServerMain
time = System.nanoTime();
randReadWrite(rate);
time = System.nanoTime() - time;
System.out.println(time/1000000000.0f);
System.err.println("cost: " + time / 1000000000.0f);
WebServerMain.threadCount --;
}
public void randReadWrite(double scale)
{
public void randReadWrite(double scale) {
Random randNum = new Random();
for (int i = 0; i < runs; i++)
{
if (Math.random()<scale)
{
for (int i = 0; i < runs; i++) {
if (Math.random() < scale) {
webSession.getInstance().get(keys[randNum.nextInt(nums)]);
}
else
{
} else {
webSession.getInstance().set(keys[randNum.nextInt(nums)], object);
}
try
{
try {
Thread.sleep((long) 0.00001);
} catch (InterruptedException e)
{
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
@ -110,9 +92,15 @@ public class WebServerMain
}
}
public static void main(String[] args)
{
PropertyConfigurator.configure(System.getProperty("user.dir")+"/bin/log4j.properties");//加载.properties文件
public static void main(String[] args) {
for (int i =0; i< 10; i++){
WebServerMain.run(args);
}
}
public static void run(String[] args) {
PropertyConfigurator.configure(System.getProperty("user.dir")
+ "/bin/log4j.properties");// 加载.properties文件
initConfig();
RegisterHandler.initHandler();
@ -125,92 +113,117 @@ public class WebServerMain
Server requestServer = Server.getInstance();
requestServer.init(8888);
int threads = Integer.parseInt(args[0]);//线程数
int runs = Integer.parseInt(args[1]); //执行次数
int Nums = Integer.parseInt(args[2]); // key数目
int size = Integer.parseInt(args[3]); // value大小
double rate = Double.parseDouble(args[4]); //读写比例
int threads = Integer.parseInt(args[0]);// 线程数
WebServerMain.threadCount = threads;
int runs = Integer.parseInt(args[1]); // 执行次数
int Nums = Integer.parseInt(args[2]); // key数目
int size = Integer.parseInt(args[3]); // value大小
double rate = Double.parseDouble(args[4]); // 读写比例
// get object to store
byte[] obj = new byte[size];
for (int i = 0; i < size; i++)
{
for (int i = 0; i < size; i++) {
obj[i] = '1';
}
String value = new String(obj);
String[] keys = new String[Nums];
for (int i = 0; i < Nums; i++)
{
for (int i = 0; i < Nums; i++) {
keys[i] = "" + i;
}
for (int i = 0; i < threads; i++)
{
for (int i = 0; i < threads; i++) {
bench b = new bench(runs, Nums, i, value, keys, rate);
b.start();
}
while (WebServerMain.threadCount != 0){
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
try {
System.out.println(webSession.results.get(0));
} catch (JSONException e1) {
e1.printStackTrace();
}
webSession.getInstance().stats();
while (webSession.stats.size() != 4 || webSession.results.length() < 3000){
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println(webSession.results.length());
// try {
// JSONObject jStats = new JSONObject(webSession.stats.get(2));
// System.out.println(jStats);
// System.out.println(jStats.get("cmd_get"));
// System.out.println(jStats.get("cmd_set"));
// } catch (JSONException e) {
// e.printStackTrace();
// }
requestServer.stop();
//webSession.session = null;
webSession.results = new JSONArray();
}
// ÁÈ¡memcached clientÅäÖÃ
public static boolean readClientsXML(String str)
{
DocumentBuilderFactory factory=DocumentBuilderFactory.newInstance();
try {
factory.setIgnoringElementContentWhitespace(true);
public static boolean readClientsXML(String str) {
DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
try {
factory.setIgnoringElementContentWhitespace(true);
DocumentBuilder db=factory.newDocumentBuilder();
Document xmldoc=db.parse(new File(str));
Element elmtInfo = xmldoc.getDocumentElement();
NodeList nodes = elmtInfo.getChildNodes();
for (int i = 0; i < nodes.getLength(); i++)
{
Node result = nodes.item(i);
if (result.getNodeType() == Node.ELEMENT_NODE && result.getNodeName().equals("client"))
{
NodeList ns = result.getChildNodes();
DocumentBuilder db = factory.newDocumentBuilder();
Document xmldoc = db.parse(new File(str));
Element elmtInfo = xmldoc.getDocumentElement();
NodeList nodes = elmtInfo.getChildNodes();
for (int i = 0; i < nodes.getLength(); i++) {
Node result = nodes.item(i);
if (result.getNodeType() == Node.ELEMENT_NODE
&& result.getNodeName().equals("client")) {
NodeList ns = result.getChildNodes();
ClientConfig localClient = new ClientConfig();
int m=0;
for (int j = 0; j < ns.getLength(); j++)
{
Node record = ns.item(j);
if (record.getNodeType() == Node.ELEMENT_NODE)
{
if (record.getNodeName().equals("id"))
{
m++;
localClient.id = Integer.decode(record.getTextContent());
}
else if (record.getNodeName().equals("host"))
{
m++;
localClient.host = record.getTextContent();
}
else if (record.getNodeName().equals("client_port"))
{
m++;
localClient.client_port = Integer.decode(record.getTextContent());
}
else if (record.getNodeName().equals("memcached"))
{
int m = 0;
for (int j = 0; j < ns.getLength(); j++) {
Node record = ns.item(j);
if (record.getNodeType() == Node.ELEMENT_NODE) {
if (record.getNodeName().equals("id")) {
m++;
localClient.id = Integer.decode(record
.getTextContent());
} else if (record.getNodeName().equals("host")) {
m++;
localClient.host = record.getTextContent();
} else if (record.getNodeName().equals(
"client_port")) {
m++;
localClient.client_port = Integer.decode(record
.getTextContent());
} else if (record.getNodeName().equals("memcached")) {
m++;
localClient.memcached = record.getTextContent();
}
}
}
}
if(m==4)
{
m_mapMemcachedClient.put(localClient.id, localClient);
}
}
}
} catch (ParserConfigurationException e) {
e.printStackTrace();
} catch (SAXException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
return true;
}
if (m == 4) {
m_mapMemcachedClient.put(localClient.id, localClient);
}
}
}
} catch (ParserConfigurationException e) {
e.printStackTrace();
} catch (SAXException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
return true;
}
}

300
MemcachedClient/src/server/webSession.java
View File

@ -1,5 +1,6 @@
package server;
package com.myself.server;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
@ -11,248 +12,168 @@ import messageBody.memcachedmsg.nm_Connected_web_back;
import messageBody.requestMsg.nr_Connected_mem_back;
import messageBody.requestMsg.nr_Read;
import messageBody.requestMsg.nr_Read_res;
import messageBody.requestMsg.nr_Stats;
import messageBody.requestMsg.nr_Stats_res;
import messageBody.requestMsg.nr_write;
import messageBody.requestMsg.nr_write_res;
import org.apache.log4j.Logger;
import org.apache.log4j.Priority;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.util.internal.ConcurrentHashMap;
import org.json.JSONArray;
import database.DBMessage;
import r_memcached.MemcachedMgr;
import server.NetMsg;
import com.myself.memcached.MemcachedMgr;
import common.EMSGID;
public class webSession implements Runnable
{
public class webSession implements Runnable {
ConcurrentLinkedQueue<MessageEvent> recvQueue = new ConcurrentLinkedQueue<MessageEvent>();
Map<Integer, Channel> MemcachedChannelMap = new ConcurrentHashMap<Integer, Channel>();
Map<Integer, Channel> RequestChannelMap = new ConcurrentHashMap<Integer, Channel>();
static webSession session = null;
public static webSession session = null;
public Channel curChannel;
public long totalTime = 0;
public long ticks =0;
public long ticks = 0;
//public static Logger log = LoggerUtil.getInstance();
public static JSONArray results = new JSONArray();
public static Map<Integer, String> stats = new HashMap<>();
// public static Logger log = LoggerUtil.getInstance();
public static Logger log = Logger.getLogger(webSession.class.getName());
public static webSession getInstance()
{
if (session == null)
{
public static webSession getInstance() {
if (session == null) {
session = new webSession();
}
return session;
}
public void start()
{
//DBSession.getInstance().start(); 数据库连接
public void start() {
// DBSession.getInstance().start(); Êý¾Ý¿âÁ¬½Ó
new Thread(session).start();
System.out.println("session start");
}
// Ìí¼ÓrequestÁ¬½Ó
public void addRequestChannel(Channel ch)
{
public void addRequestChannel(Channel ch) {
RequestChannelMap.put(ch.getId(), ch);
}
// ɾ³ýrequestÁ¬½Ó
public void removeRequestChannel(Channel ch)
{
public void removeRequestChannel(Channel ch) {
RequestChannelMap.remove(ch.getId());
}
// »ñµÃrequestÁ¬½Ó
public Channel getRequestChannel(Integer id)
{
public Channel getRequestChannel(Integer id) {
return RequestChannelMap.get(id);
}
// Ôö¼ÓclientÁ¬½Ó
public void addClientChannel(Integer num,Channel ch)
{
public void addClientChannel(Integer num, Channel ch) {
MemcachedChannelMap.put(num, ch);
}
public Channel getClientChannel(Integer id)
{
public Channel getClientChannel(Integer id) {
return MemcachedChannelMap.get(id);
}
// ɾµôclientÁ¬½Ó
@SuppressWarnings("rawtypes")
public void removeClientChannel(Channel ch)
{
public void removeClientChannel(Channel ch) {
Iterator iter = MemcachedChannelMap.entrySet().iterator();
while (iter.hasNext())
{
while (iter.hasNext()) {
Entry entry = (Entry) iter.next();
if ((Channel)entry.getValue() == ch)
{
MemcachedChannelMap.remove((Integer)entry.getKey());
if ((Channel) entry.getValue() == ch) {
MemcachedChannelMap.remove((Integer) entry.getKey());
break;
}
}
}
//////////////////////////////////////////////////////////
public void run()
{
while(true)
{
// ////////////////////////////////////////////////////////
public void run() {
while (true) {
MessageEvent event = recvQueue.poll();
while(event != null)
{
while (event != null) {
handle(event);
event = recvQueue.poll();
}
try
{
try {
Thread.sleep(0);
} catch (InterruptedException e)
{
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
public int gethashMem(String key)
{
return Math.abs(key.hashCode()%MemcachedMgr.getInstance().getSize());
public int gethashMem(String key) {
return Math.abs(key.hashCode() % MemcachedMgr.getInstance().getSize());
}
public void handle(MessageEvent e)
{
NetMsg msg = (NetMsg)e.getMessage();
switch (msg.getMsgID())
{
case nm_connected:
{
public void handle(MessageEvent e) {
NetMsg msg = (NetMsg) e.getMessage();
switch (msg.getMsgID()) {
case nm_connected: {
nm_Connected msgBody = msg.getMessageLite();
addClientChannel(msgBody.getNum(), e.getChannel());
nm_Connected_web_back.Builder builder = nm_Connected_web_back.newBuilder();
NetMsg send = NetMsg.newMessage();
send.setMessageLite(builder);
send.setMsgID(EMSGID.nm_connected_web_back);
sendMsg(e.getChannel(), send);
}
break;
case nr_connected_mem_back:
{
nr_Connected_mem_back msgLite= msg.getMessageLite();
break;
case nr_connected_mem_back: {
nr_Connected_mem_back msgLite = msg.getMessageLite();
addClientChannel(msgLite.getMemID(), e.getChannel());
}
break;
case nr_read_res:
{
case nr_stats_res: {
nr_Stats_res msgBody = msg.getMessageLite();
stats.put(msg.getNodeRoute(), msgBody.getValue());
//System.out.println("stats:\n" + msgBody.getValue());
}
break;
case nr_read_res: {
nr_Read_res msgBody = msg.getMessageLite();
// System.out.println("key:"+msgBody.getKey()+" value:"+msgBody.getValue());
// System.out.println(String.valueOf((System.nanoTime()-msgBody.getTime())/1000000.0));
// log.log(Priority.INFO, String.valueOf((System.nanoTime()-msgBody.getTime())/1000000.0));
// log.log(Level.INFO, String.valueOf((System.nanoTime()-msgBody.getTime())/1000000.0));
// System.err.println((System.nanoTime()-msgBody.getTime())/1000000.0);
// if (ticks==0)
// {
// totalTime = System.currentTimeMillis();
// }
// totalTime += System.nanoTime()-msgBody.getTime();
// ticks++;
// if (ticks == 1000)
// {
// System.out.println(totalTime/1000000000.0f);
//
// totalTime = 0;
// ticks = 0;
// }
// if (msgBody.getValue().isEmpty()) //读数据库
// {
// DBMessage dbMsg=new DBMessage();
// dbMsg.mode = DBMessage.mode_query;
// dbMsg.key = msgBody.getKey();
//
// DBSession.getInstance().addDBMessage(dbMsg);
// }
// else
// {
// nc_ReadRes.Builder builder = nc_ReadRes.newBuilder();
// builder.setKey(msgBody.getKey());
// builder.setValue(msgBody.getValue());
//
// NetMsg send = NetMsg.newMessage();
// send.setMessageLite(builder);
// send.setMsgID(EMSGID.nc_read_res);
//
// //sendMsg(getRequestChannel(msgBody.getClientid()), send);
// }
Map<String, String> readResult = new HashMap<String, String>();
readResult.put("type", "GET");
readResult.put("node", ""+msg.getNodeRoute());
readResult.put("key", msgBody.getKey());
readResult.put("value", msgBody.getValue());
results.put(readResult);
//System.out.println("key:"+msgBody.getKey()+" value:"+msgBody.getValue());
}
break;
case nr_write_res:
{
break;
case nr_write_res: {
nr_write_res msgBody = msg.getMessageLite();
// System.out.println("key:"+msgBody.getKey()+" value:"+msgBody.getValue());
//DBMessage dbMsg = new DBMessage(); //异步写数据库
//dbMsg.mode = DBMessage.mode_set;
//dbMsg.key = msgBody.getKey();
//dbMsg.value = msgBody.getValue();
//DBSession.getInstance().addDBMessage(dbMsg);
// log.log(Level.INFO, String.valueOf((System.nanoTime()-msgBody.getTime())/1000000.0));
//System.err.println();
// System.out.println(String.valueOf((System.nanoTime()-msgBody.getTime())/1000000.0));
//
// if (ticks==0)
// {
// totalTime = System.currentTimeMillis();
// }
// totalTime += System.nanoTime()-msgBody.getTime();
// ticks++;
// if (ticks == 1000)
// {
// System.out.println(totalTime/1000000000.0f);
//
// totalTime = 0;
// ticks = 0;
// }
// nc_WriteRes.Builder builder = nc_WriteRes.newBuilder();
// builder.setKey(msgBody.getKey());
// builder.setValue(msgBody.getValue());
//
// NetMsg send = NetMsg.newMessage();
// send.setMessageLite(builder);
// send.setMsgID(EMSGID.nc_write_res);
//sendMsg(getRequestChannel(msgBody.getClientid()), send);
Map<String, String> readResult = new HashMap<String, String>();
readResult.put("type", "SET");
readResult.put("node", ""+msg.getNodeRoute());
readResult.put("key", msgBody.getKey());
readResult.put("value", msgBody.getValue());
results.put(readResult);
//System.out.println("key:"+msgBody.getKey()+" value:"+msgBody.getValue());
}
break;
break;
default:
System.err.println(msg.getMsgID().toString());
break;
}
//log.log(Level.INFO, msg.getMsgID().toString());
}
public void addSession(MessageEvent e)
{
public void addSession(MessageEvent e) {
recvQueue.offer(e);
}
public boolean sendAllMsg(Integer hash, NetMsg msg)
{
for (int i = 0; i < MemcachedMgr.nCopyNode; i++)
{
Channel eChannel = getClientChannel(hash+i);
if (eChannel != null)
{
public boolean sendAllMsg(Integer hash, NetMsg msg) {
for (int i = 0; i < MemcachedMgr.nCopyNode; i++) {
Channel eChannel = getClientChannel(hash + i);
if (eChannel != null) {
sendMsg(eChannel, msg);
return false;
}
@ -260,32 +181,40 @@ public class webSession implements Runnable
return true;
}
public boolean randSendMsg2Memcached(Integer hash, NetMsg msg)
{
public boolean allSendMsg2Memcached(NetMsg msg) {
int size = MemcachedChannelMap.size();
for (int i = 0; i < size; i++) {
Channel eChannel = getClientChannel(i);
if (eChannel != null) {
sendMsg(eChannel, msg);
}
}
return true;
}
public boolean randSendMsg2Memcached(Integer hash, NetMsg msg) {
Random random = new Random();
int index = random.nextInt(MemcachedMgr.nCopyNode);
for (int i = 0; i < MemcachedMgr.nCopyNode; i++)
{
int num = (hash+i+index+MemcachedMgr.getInstance().getSize())%MemcachedMgr.getInstance().getSize();
for (int i = 0; i < MemcachedMgr.nCopyNode; i++) {
int num = (hash + i + index + MemcachedMgr.getInstance().getSize())
% MemcachedMgr.getInstance().getSize();
Channel eChannel = getClientChannel(num);
if (eChannel != null)
{
if (eChannel != null) {
sendMsg(eChannel, msg);
return true;
}
}
System.err.println("SendMsg wrong : randSendMsg2Memcached in webSession line 174");
System.exit(-1);
return false;
}
public boolean SendMsg2Leader(Integer hash, NetMsg msg)
{
for (int i = 0; i < MemcachedMgr.nCopyNode; i++)
{
int index = (hash+i+MemcachedMgr.getInstance().getSize())
%MemcachedMgr.getInstance().getSize();
public boolean SendMsg2Leader(Integer hash, NetMsg msg) {
for (int i = 0; i < MemcachedMgr.nCopyNode; i++) {
int index = (hash + i + MemcachedMgr.getInstance().getSize())
% MemcachedMgr.getInstance().getSize();
Channel eChannel = getClientChannel(index);
if (eChannel != null)
{
if (eChannel != null) {
sendMsg(eChannel, msg);
return true;
}
@ -293,21 +222,25 @@ public class webSession implements Runnable
return false;
}
public void sendMsg(Channel ch, NetMsg msg)
{
try
{
public void sendMsg(Channel ch, NetMsg msg) {
try {
ch.write(msg);
}
catch (Throwable e)
{
//log.log(Level.WARNING, "send msg fail");
} catch (Throwable e) {
// log.log(Level.WARNING, "send msg fail");
}
}
public boolean stats() {
nr_Stats.Builder builder = nr_Stats.newBuilder();
builder.setTime(System.nanoTime());
NetMsg msg = NetMsg.newMessage();
msg.setMessageLite(builder);
msg.setMsgID(EMSGID.nr_stats);
allSendMsg2Memcached(msg);
return true;
}
public boolean get(String key)
{
public boolean get(String key) {
nr_Read.Builder builder = nr_Read.newBuilder();
builder.setKey(key);
builder.setTime(System.nanoTime());
@ -318,8 +251,7 @@ public class webSession implements Runnable
return true;
}
public boolean set(String key, String value)
{
public boolean set(String key, String value) {
nr_write.Builder builder = nr_write.newBuilder();
builder.setKey(key);
builder.setValue(value);