进入idea，新建一个maven项目
主要是模拟150个设备同时并发，并发时间持续15min
1.创建客户端，构造请求发送到对应的rabbitmq的队列，用的protobuf协议。
1import com.google.protobuf.ByteString; 2import com.rabbitmq.client.*; 3import org.apache.commons.codec.DecoderException; 4import org.apache.commons.codec.binary.Hex; 5import java.io.IOException; 6import java.text.SimpleDateFormat; 7import java.util.Date; 8import java.util.concurrent.TimeoutException; 9 10public class Producer { 11 12//上传的命令字段值参数化 13private final static String[] COMMANDS = new String[]{ 14 15 "XXXXX", "YYYYYY" "ZZZZZ", 16}; 17 18 private int index; 19 20 public Producer(int index) { 21 this.index = index; 22 } 23 24 public byte[] message(byte[] command) { 25 //根据proto文本消息生成的slot3编辑脚本，构建一个消息 26 Slot3.SlotMessage.Builder slots = Slot3.SlotMessage.newBuilder(); 27 slots.setOpenId("XXX"); 28 slots.setProductId("YYYY"); 29 //长整型 30 String NO = String.valueOf(70000000000l + index); 31 slots.setNodeEui(NO); 32 slots.setCommand(Slot3.SlotMessage.Command.DOWNLINK); 33 SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); 34 Date date = new Date(); 35 String time1 = f.format(date); 36 slots.setTimestamp(time1); 37 Slot3.SlotMessage.Payload.Builder payload = Slot3.SlotMessage.Payload.newBuilder(); 38 ByteString command2 = ByteString.copyFrom(command); 39 payload.setData(command2); 40 slots.addPayload(payload); 41 slots.setAppMessageType(0); 42 Slot3.SlotMessage msg = slots.build(); 43 System.out.println("before:" + msg); 44 45 System.out.println("===msg Byte:"); 46 byte[] msgbyteArray = msg.toByteArray(); 47 System.out.println(msgbyteArray); 48 return msgbyteArray; 49 } 50 51 /** 52 * 构造函数 53 * RabbitMQ客户端相关配置 54 * 连接AMQP broker并初始化队列 55 */ 56 public void produce() throws IOException, TimeoutException { 57 58 ConnectionFactory factory = new ConnectionFactory(); 59 //设置RabbitMQ相关信息 60 factory.setHost("10.10.XX.XX"); 61 factory.setPort(5671); 62 factory.setUsername("rabbitmq用户名"); 63 factory.setPassword("rabbitmq密码"); 64 factory.setVirtualHost("/"); 65 //创建发送消息rabbitmq信息的连接 66 Connection connection = factory.newConnection(); 67 Channel channel = connection.createChannel(); 68 69 String exchange = "交互机名称"; 70 String queue = "队列名称"; 71 72 for (String input : COMMANDS) { 73 byte[] body = message(decodeHex(input)); 74 75 channel.basicPublish(exchange, queue, null, body); 76 77 try { 78//生成0~1秒的随机 79// Double code = (Math.random() * 9 + 1) * 100; 80// Long ms =code.longValue(); 81//每个命令请求间隔1s 82 Thread.sleep(1000l); 83 } catch (Exception e) { 84 e.printStackTrace(); 85 } 86 87 } 88 89 channel.close(); 90 connection.close(); 91 92 93 } 94 95 /** 96 * Hex解码. 97 */ 98 public static byte[] decodeHex(String input) { 99 try { 100 return Hex.decodeHex(input.toCharArray()); 101 } catch (DecoderException e) { 102 throw new RuntimeException(e); 103 } 104 } 105 106 107}
View Code
2.模拟150个设备同时并发，发送消息
1 1 public class ConsumerClient { 2 2 3 3 public static void main(String[] args) throws Exception { 4 4 //并发150次 5 5 for (int i = 0; i < 150; i++) { 6 6 final int index = i; 7 7 new Thread(() -> { 8 8 //创建producer 9 9 Producer producer = new Producer(index); 1010 Long start = System.currentTimeMillis(); 1111 1212 while (true) { 1313 try { 1414 //发送消息 1515 producer.produce(); 1616 } catch (Exception e) { 1717 e.printStackTrace(); 1818 } 1919 //持续运行15分钟 2020 Long end = System.currentTimeMillis(); 2121 if (end - start >= 15 * 60 * 1000l) { 2222 break; 2323 } 2424 } 2525 2626 }).start(); 2727 2828 } 2929 } 3030 }
View Code
3.pom.xml
1<?xml version="1.0" encoding="UTF-8"?> 2<project xmlns="http://maven.apache.org/POM/4.0.0" 3 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 4 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 5 <modelVersion>4.0.0</modelVersion> 6 7 <groupId>Commandxingneng</groupId> 8 <artifactId>Commandxingneng</artifactId> 9 <version>1.0-SNAPSHOT</version> 10 <build> 11 <plugins> 12 <plugin> 13 <groupId>org.apache.maven.plugins</groupId> 14 <artifactId>maven-compiler-plugin</artifactId> 15 <configuration> 16 <source>1.8</source> 17 <target>1.8</target> 18 </configuration> 19 </plugin> 20 </plugins> 21 </build> 22 23 <dependencies> 24 <dependency> 25 <groupId>mysql</groupId> 26 <artifactId>mysql-connector-java</artifactId> 27 <version>5.1.26</version> 28 <scope>test</scope> 29 </dependency> 30 <dependency> 31 <groupId>com.rabbitmq</groupId> 32 <artifactId>amqp-client</artifactId> 33 <version>3.6.5</version> 34 </dependency> 35 <dependency> 36 <groupId>com.google.protobuf</groupId> 37 <artifactId>protobuf-java</artifactId> 38 <version>3.6.1</version> 39 </dependency> 40 <dependency> 41 <groupId>io.vertx</groupId> 42 <artifactId>vertx-core</artifactId> 43 <version>RELEASE</version> 44 </dependency> 45 <dependency> 46 <groupId>commons-codec</groupId> 47 <artifactId>commons-codec</artifactId> 48 <version>RELEASE</version> 49 </dependency> 50 <dependency> 51 <groupId>commons-codec</groupId> 52 <artifactId>commons-codec</artifactId> 53 <version>RELEASE</version> 54 </dependency> 55 </dependencies> 56 57</project>
View Code
4.运行ConsumerClient.java就好了
5.proto3文本内容，怎么生成编辑脚本，参考python模拟上报消息到rabbitMQ（protobuf） ，java一样
syntax = "proto3";
message SlotMessage
{
string openId = 1;
string productId = 2;
string nodeEui = 3;
Command command = 4;
string timestamp = 5;
bool encrypted = 6;
repeated Payload payload = 7;
int32 appMessageType = 8;
enum Command {
U = 0;
D = 1;
C = 2;
O_RESULT = 3;
}
message Payload {
bytes data = 1;
Connect connect = 2;
repeated OResult oResult = 3;
enum Connect{
OFF = 0;
ON = 1;
HE = 2;
}
message OResult {
string d = 1;
int32 r = 2;
string s = 3;
}
}
}
6.查看web页面的rabbitmq消息处理情况，是否有阻塞
7.用nmon工具监控被压测的服务器，需要安装和服务器版本相对应的nmon版本
./nmon -c 120 -s 10 -f
-f ：按标准格式输出文件名称 生成文件：<hostname>_YYYYMMDD_HHMM.nmon
-t ： 输出最耗资源的进程
-s ：每隔n秒采集一次，这里为10秒
-c ：采集次数，这里为90，即监控=10*120/60=20分钟
ps -ef | grep nmon #查询nmon进程
kill -9 进程ID #强行中断监控进程
nmon文件转换
sort localhost_170616_0138.nmon>localhost_170616_0138.csv
最后从服务器中导出csv文件，用nmon analyser v55 工具分析；