JMeter中利用Jython运行Python代码

news2024/9/26 1:26:50

介绍

Jython是Python和Java的结合。Jython语法和Python一样，不但可以使用Python的库，而且还可以调用Java的库。结合了Python和Java的优点，也就是说Jython既有动态语言的灵活性，又可以用静态语言的强大的类库。其实，我理解 Jython 说简单一些，就是用 Java 语言实现的 Python 解释器，这种关系也就意味着你可以用 Python 语言编写程序而同时使用 Java 库。

Jython使用Python2编写，可以在Beanshell中直接编写Python代码，不需要安装Python2解析器。

JMmter中的使用

官网 / 使用手册

jmeter运行Python的依赖包：jython-standalone-2.7.2.jar 下载完成后放到jmeter目录下的lib/ext 下，重启jmeter。

Jython中包含的包

Jython中含有的python包：

https://github.com/jython/jython/tree/master/lib-python/2.7

https://github.com/jython/jython/tree/master/Lib

在 Jython 中，导入语句是一个可以出现在源文件中任何位置的表达式，甚至可以有条件地执行。

外部包导入

Python：同名的包，导入的第一个 python 模块或包会阻止其他模块或包的导入。

Java包：都可以导入；

jython只支持python2，如若导入python3的包，部分会出现不兼容的情况(兼容部分是可以使用的)；

# 导入第三方包示例
import sys
sys.path.append('C:\Users\laiqu\AppData\Local\Programs\Python\Python27\Lib')
sys.path.append('C:\Users\laiqu\AppData\Local\Programs\Python\Python27\Lib\site-packages')

应用示例

示例1：JMeter中使用Python进行断言

import sys
reload(sys)
#保证结果树的响应数据中的中文不为乱码，否则会提示错误信息：jmeter python Cannot create PyString with non-byte value
sys.setdefaultencoding('utf8')  


# 获取msg变量
msg = vars.get("msg")
# 打印msg
log.info(msg)
# 判断msg 是否等success，是则断言通过，否则断言失败
if msg == "success":
    AssertionResult.setFailure(False)
else:
    AssertionResult.setFailure(True)

示例2：JMeter中运行Python数据处理

import sys
reload(sys)
sys.setdefaultencoding('utf8')      #保证结果树的响应数据中的中文不为乱码，否则会提示错误信息：jmeter python Cannot create PyString with non-byte value

# python脚本：将jmeter的变量赋值到python,然后对其进行排序，并输出排序后的字符串。
a = vars.get("b")                         #jmeter的变量b赋值到python
print ("将a排序显示，办法1（列表）：")
lista = list(a)                                 #字符串转为列表
lista.sort()                                    #列表排序
bb = str(lista)
newa = ''                                      #创建一个空字符串
for i in lista:
     newa= newa+i                        #排序后的列表每项加给新字符串
vars.put("newaj",newa);              #排序后的字符串赋值给jmeter的变量newaj
c = vars.get('newaj')                    #将newaj的值给变量c
SampleResult.setResponseData("原字符串："+"${b}"+'\n'+"排序后的字符串使用vars.get打印出："+vars.get('newaj')+'\n'+"or直接使用变量打印出："+c);      
# 响应信息输出，可直接使用vars.get('newaj') 输出，也可使用变量vars.get('newaj') 输出。
SampleResult.setSuccessful(True)
SampleResult.setResponseCode("200")

#获取jmeter 参数
param =vars.get("param")
print "hello"
# 设置响应信息
SampleResult.setResponseData("{'msg':'success'}");
SampleResult.setResponseCode("200")
SampleResult.setSuccessful(True)

示例3：JMeter中连接Mysql

import sys
reload(sys)
sys.setdefaultencoding('utf8')     #设置默认编码方式
from com.ziclix.python.sql import zxJDBC

DADABASE = {
    'NAME': '',  # 数据库实例名
    'USER': '',  # 用户名
    'PASSWORD': '',  # 密码
    'HOST': 'mysql-sit.shenlanbao.com',  # 服务器ID（或域名）
    'PORT': '3306',  # 端口号
}

def select(query_sql):
    driver = "com.mysql.jdbc.Driver"
    url = "jdbc:mysql://t8888service"
    conn = zxJDBC.connect(url , DADABASE["USER"] , DADABASE["PASSWORD"] , driver)
    cur = conn.cursor()
    cur.execute(query_sql)
    data = cur.fetchall()
    conn.close()
    return data

data = select(query_sql="select * from customer where id < 10")

for i in data:
    log.info("---------------"+str(i))

封装成类

import sys

reload(sys)
sys.setdefaultencoding('utf-8')  # 设置默认编码方式
from com.ziclix.python.sql import zxJDBC

class Db():
    def __init__(self):
        url = "jdbc:mysql://mysql.com/consult-service?useUnicode=true&characterEncoding=utf8"
        self.conn = zxJDBC.connect(url, "username", "password", "com.mysql.jdbc.Driver", charset='utf-8')
        self.cur = self.conn.cursor()
        
    def __del__(self):
        self.conn.close()
        
    def close(self):
        self.__del__()
        
    def select(self, query_sql):
        self.cur.execute(query_sql)
        # 把结果转成dict返回（时间格式有点问题）
        dict_list = [dict(zip([key[0] for key in self.cur.description], values)) for values in self.cur.fetchall()]
        return dict_list
        
    def write(self, sql):
        dict_list = self.cur.execute(sql)
        self.conn.commit()
        return dict_list


db = Db()

query_sql = "select * from customer where id <10"
dict_list = db.select(query_sql = query_sql)
log.info("---------------" + str(type(dict_list)))
for i in dict_list:
    log.info("---------------" + i['phone_operator'])

query_sql = "update customer set gender = 1 where id = 1;"
w = db.write(query_sql)
log.info("w2wwwwwwwwwwww"+str(w));

在Python 2中，reload(sys)函数用于重新加载sys模块。sys模块是用于与Python解释器进行交互的系统特定参数和函数的集合。通过重新加载sys模块，你可以更新sys模块中的变量和函数。

在Python 3中，reload()函数已经被移除，可以通过importlib模块中的reload()函数来实现相同的功能。

通常大多数人执行reload(sys)这条语句其实仅仅是为了能够修改Python的默认字符集，也就是能够调用sys.setdefaultencoding()。python的默认的编码是ascii，当程序中出现非ascii编码时，python会报错UnicodeDecodeError: ‘ascii’ codec can’t decode byte 0x?? in position 1: ordinal not in range(128)

zxJDBC.connect( *, *, *, "com.mysql.jdbc.Driver", charset='utf-8')，不同数据库类型，参考JDBC连接池写就好。

示例4：引用Java包与方法

from com.tool import JsonUtil
// JsonUtil 是Java包
jsonutil = JsonUtil()

responseString = prev.getResponseDataAsString();
getJsonPathDate = jsonutil.ParaseJson(responseString,"$..token")
log.info("------------"+getJsonPathDate);

导入说明