1：关于smack与tigase的用法跟作用请大家自己去网上查看相关资料，这里就不做描述了。
PS：这篇文章主要是说明在客户端jvm创建的最大线程数的大小。
之前公司要求做一个客户端用于测试刚刚部署的tigase的性能，所以项目经理就安排了一个事情就是自己动手在客户端写一个基于smack长连接的压力测试工具。
初期的要求是这样的：
1、并发注册10000个用户
2、用户之间相互收发消息
3、可以调整并发数量
用smack 做客户端连接
由于之前对多线程这块懂的不是很深，在这个过程中碰了走了很多弯路(总以为是代码的问题，实际上不是)，我的机子的配置是：32位，xp，4G内存。自己写了个程序无论怎么跑在线数都不能突破2000,离要求还差一大截呢！自己反反复复检查程序，没啥不对啊！后来使用jconsole工具观察发现线程数6k一直是最高峰，难怪用户量上不去，要知道要保证10000个用户同时在线至少客户端要启动10000个线程吧！问题总算找到了，于是百度了一下，发现了一个jvm启动线程数的公式：
(MaxProcessMemory - JVMMemory - ReservedOsMemory) / (ThreadStackSize) = Number of threads
MaxProcessMemory 指的是一个进程的最大内存
JVMMemory JVM内存
ReservedOsMemory 保留的操作系统内存
ThreadStackSize 线程栈的大小
原来：在java语言里， 当你创建一个线程的时候，虚拟机会在JVM内存创建一个Thread对象同时创建一个操作系统线程，而这个系统线程的内存用的不是JVMMemory，而是系统中剩下的内存(MaxProcessMemory - JVMMemory - ReservedOsMemory)
OK问题找到了有了解决问题的思路：根据自己的测试发现当给jvm设置的内存数为512m时创建的线程数是最多的,如下：
2*1024*1024-1024*64-512*1024)/128=11776
又通过观察发现要保持一个长连接本地至少了启动四个线程于是支持的在线用户数为:11776/4
OK:测试结果想如下：
看图说话，哈哈 问题解决：
代码如下：
public class SmackConf {
public static String server="XXXX";
public static int port=5222;
private SmackConf(){};
//初始化配置
private static class ConnectionConfig{
private static ConnectionConfiguration config;
static{
config = new ConnectionConfiguration(server, 5222, "XXX");
config.setSASLAuthenticationEnabled(false);
config.setSecurityMode(ConnectionConfiguration.SecurityMode.disabled);
//config.setSendPresence(false);
System.out.println("加载配置文件类完毕");
}
}
public static ConnectionConfiguration getConnectionConfiguration(){
return ConnectionConfig.config;
}
}
public class SmackUtils {
private static Logger log = Logger.getLogger(SmackUtils.class);
private static int sleepTime = 1000;
public static void chart(User user) {
Connection connection = MapsUtils.getConnection(user.getUsername());
try {
if(connection==null||!connection.isConnected()){
initConnection(user);
}
if (!connection.isAuthenticated()) {
initLogin(user);
}
synchronized (connection) {
PacketFilter filter = new MessageTypeFilter(Message.Type.chat);
connection.addPacketListener(new PacketListener() {
@Override
public void processPacket(Packet packet) {
Message message = (Message) packet;
if (message.getBody() != null) {
String fromName = (String) message
.getProperty("fromId");
System.out.println("Got text [" + message.getBody()
+ "] from [" + fromName + "]");
}
}
}, filter);
}
} catch (Exception e) {
System.out.println("chart===="+e.getMessage());
log.info("chart===="+e.getMessage());
}
}
/*
* 获得连接
*/
private static void getConnection(User user, int i) throws Exception {
Connection connection = null;
try {
log.info("第" + i + "次开始申请连接，用户：" + user.getUsername());
connection = MapsUtils.getConnection(user.getUsername());
if (connection == null) {
connection = new XMPPConnection(SmackConf.getConnectionConfiguration());
MapsUtils.put(user.getUsername(), connection);
}
if(connection.isConnected()){
connection.disconnect();
}
connection.connect();
user.setConnectionFa(true);
log.info("第" + i + "次申请连接，用户：" + user.getUsername() + "成功");
} catch (Exception e) {
log.info("第" + i + "次申请连接，用户：" + user.getUsername() + "失败");
System.out.println("获得连接的err:"+e.getMessage());
log.info("获得连接的err:"+e.getMessage());
throw e;
}
}
public static void getConnection(User user) {
/** 建立连接 */
try {
getConnection(user, 0);
} catch (Exception e) {
while (user.getiConnent() < MapsUtils.count
&& !user.isConnectionFa()) {
try {
Thread.sleep(sleepTime);
} catch (InterruptedException e1) {
}
user.setiConnent(user.getiConnent() + 1);
try {
getConnection(user, user.getiConnent());
} catch (Exception e1) {
e1.printStackTrace();
}
}
}
}
private static void initConnection(User user) throws Exception {
try {
log.info("开始初始化创建连接：" + user.getUsername());
user.setConnectionFa(false);
user.setiConnent(0);
getConnection(user);
} catch (Exception e) {
throw e;
}
}
/**
* 注册新用户
*
* @param i
* @throws Exception
*/
private static void createAccount(User user, int i) throws Exception {
Connection connection = MapsUtils.getConnection(user.getUsername());
try {
if (connection == null || !connection.isConnected()) {
initConnection(user);
}
synchronized (connection) {
AccountManager am = null;
log.info("第" + i + "次开始注册用户：" + user.getUsername());
am = connection.getAccountManager();
am.createAccount(user.getUsername(), user.getPassword());
user.setCreateFa(true);
log.info("第" + i + "次开始注册用户：" + user.getUsername() + "成功");
}
} catch (Exception e) {
log.info("第" + i + "次开始注册用户：" + user.getUsername() + "失败");
String message=e.getMessage();
log.info("注册用户的err:"+e.getMessage());
reInit(message, user);
throw e;
}
}
public static void createAccount(User user) {
try {
createAccount(user, 0);
} catch (Exception e) {
while (user.getiCreate() < MapsUtils.count && !user.isCreateFa()) {
try {
Thread.sleep(sleepTime);
} catch (InterruptedException e2) {
}
user.setiCreate(user.getiCreate() + 1);
try {
createAccount(user, user.getiCreate());
} catch (Exception e1) {
e1.printStackTrace();
}
}
}
}
// 登陆
public static void login(User user) {
try {
login(user, 0);
} catch (Exception e) {
while (user.getiLogin() < MapsUtils.count && !user.isLoginFa()) {
try {
Thread.sleep(sleepTime);
} catch (InterruptedException e1) {
}
user.setiLogin(user.getiLogin() + 1);
try {
login(user, user.getiLogin());
} catch (Exception e2) {
e2.printStackTrace();
}
}
}
}
private static void initSignUser(User user) throws Exception {
try {
log.info("开始初始化注册：" + user.getUsername());
user.setiCreate(0);
user.setCreateFa(false);
createAccount(user);
} catch (Exception e) {
throw e;
}
}
// 登陆
private static void login(User user, int i) throws Exception {
Connection connection = MapsUtils.getConnection(user.getUsername());
try {
if(connection==null){
initConnection(user);
}
if(!connection.isConnected()){
initConnection(user);
}
if (!user.isCreateFa()) {
initSignUser(user);
}
if(connection.isAuthenticated()){
return;
}
synchronized (connection) {
log.info("第" + i + "次开始登陆，用户：" + user.getUsername());
connection.login(user.getUsername(), user.getPassword());
user.setLoginFa(true);
/** 设置状态 */
Presence presence = new Presence(Presence.Type.available);
presence.setStatus("Q我吧");
connection.sendPacket(presence);
log.info("第" + i + "次开始登陆，用户：" + user.getUsername() + "成功");
}
} catch (Exception e) {
log.info("第" + i + "次开始登陆，用户：" + user.getUsername() + "失败");
String message = e.getMessage();
log.error("登陆异常信息:" + message);
System.out.println("用户登陆的err:"+e.getMessage());
log.info("用户登陆的err:"+e.getMessage());
reInit(message, user);
throw e;
}
}
private static void initLogin(User user) throws Exception {
try {
log.info("开始初始化登陆：" + user.getUsername());
user.setiLogin(0);
user.setLoginFa(false);
login(user);
} catch (Exception e) {
throw e;
}
}
public static void sendMessage(String from, String to, String message) {
/** 获取当前登陆用户的聊天管理器 */
int messageIndex = 0;
try {
sendMessage(from, to, message, messageIndex);
} catch (Exception e) {
e.printStackTrace();
while (messageIndex < MapsUtils.count) {
System.err.println("消息发送次数：" + messageIndex + "消息来源：" + from
+ "\t消息目的地:" + to + "\t消息内容:" + message);
try {
Thread.sleep(sleepTime);
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
messageIndex++;
try {
sendMessage(from, to, message, messageIndex);
} catch (Exception e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
}
}
}
private static void sendMessage(String from, String to, String message,
int count) throws Exception {
/** 获取当前登陆用户的聊天管理器 */
Connection connection = MapsUtils.getConnection(from);
try {
if (!connection.isAuthenticated()) {
log.info(from + "==没有登陆");
return;
}
synchronized (connection) {
/** 发送消息 */
ChatManager chatManager = connection.getChatManager();
Chat chat = chatManager.createChat(to + "@tt.com", null);
chat.sendMessage("消息来源：" + from + "\t消息目的地:" + to + "\t消息内容:"
+ message);
System.err.println("消息来源：" + from + "\t消息目的地:" + to + "\t消息内容:"
+ message);
}
} catch (Exception e) {
throw e;
}
}
private static void reInit(String message,User user) throws Exception{
if(message==null){
throw new IllegalArgumentException("message参数不能为空");
}
try {
if(message.indexOf("Not connected")>-1){
//没有连接
initConnection(user);
}else if(message.indexOf("not-authorized(401)")>-1){
//没有注册
initSignUser(user);
}else if(message.indexOf("conflict(409)")>-1){
//重复注册
user.setCreateFa(true);
}else if(message.indexOf("No response")>-1){
//没有返回
initConnection(user);
}
} catch (Exception e) {
throw e;
}
}
}
private ExecutorService executorService=null;
public ThreadSysUilt(ExecutorService executorService) {
super();
this.executorService = executorService;
}
/**
* 执行线程任务
* @param rs
* @param fa true(代表全部线程任务需要执行完才能返回)，false(与true相反)
* @return
*/
public void execute(List<Runnable> rs){
for(Runnable r:rs){
executorService.execute(r);
}
executorService.shutdown();
}
/**
* 线程执行任务
* @param cs
* @return
*/
public <V> List<Future<V>> submit(List<Callable<V>> cs){
List<Future<V>> fLists=null;
try {
fLists = executorService.invokeAll(cs);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
executorService.shutdown();
return fLists;
}
public class ConnThreadMore implements Callable<MoreUser> {
private Logger log = Logger.getLogger(ConnThreadSpe.class.getName());
private MoreUser moreUser = null;
public ConnThreadMore(MoreUser moreUser) {
super();
this.moreUser = moreUser;
}
@Override
public MoreUser call() {
// TODO Auto-generated method stub
List<User> users=moreUser.getUsers();
for(User user:users){
log.info("开始新建线程:"+Thread.currentThread().getName()+"==执行："+user.getUsername()+"==创建连接");
try {
SmackUtils.getConnection(user);
Thread.sleep(1000);
} catch (Exception e) {
log.error("申请连接异常", e);
}
try {
SmackUtils.createAccount(user);
Thread.sleep(1000);
} catch (Exception e) {
log.error("申请连接异常", e);
}
}
return moreUser;
}
}
public class LoginThreadMore implements Runnable {
private Logger log = Logger.getLogger(LoginThreadSpe.class.getName());
private MoreUser moreUser = null;
public LoginThreadMore(MoreUser moreUser) {
super();
this.moreUser = moreUser;
}
@Override
public void run() {
// TODO Auto-generated method stub
List<User> users=moreUser.getUsers();
for(User user:users){
log.info("开始新建线程:"+Thread.currentThread().getName()+"==执行："+user.getUsername()+"==注册并登录");
try {
SmackUtils.login(user);//登录
Thread.sleep(100);
} catch (InterruptedException e) {
log.info("登陆异常", e);
e.printStackTrace();
}
try {
SmackUtils.chart(user);//启动消息
Thread.sleep(100);
} catch (InterruptedException e) {
log.info("启动消息", e);
e.printStackTrace();
}
}
}
}
总结：这个实现上是比较简单的，关键的问题是培养自己解决问题的能力，有不对之处希望大家多多指正。
使用smack对tigase进行压力测试,布布扣,bubuko.com