JPOS mux 上的多个 ISO8583 请求并行
Multiple ISO8583 request parallel on a JPOS mux
我有一个可以异步处理请求的工作 JPOS ISO8583 服务器,然后我需要开发一个测试客户端来对其进行压力测试。
我想知道,多路复用器是否能够同时处理多个请求并并行处理它?请参阅此图。我正在使用 ISOMUX(不是 QMUX)。
如果答案是肯定的,那怎么办?
我试过这个代码:
Customizer.java
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import org.jpos.iso.ISODate;
import org.jpos.iso.ISOException;
import org.jpos.iso.ISOMUX;
import org.jpos.iso.ISOMsg;
import org.jpos.iso.channel.ASCIIChannel;
import org.jpos.iso.packager.ISO87APackager;
public class Customizer {
private ISO87APackager packager;
private ASCIIChannel channel;
private ISOMUX mux;
public Customizer() {
try {
this.packager = new ISO87APackager();
this.channel = new ASCIIChannel(packager);
} catch (IOException e) {
e.printStackTrace();
}
}
static String C_HOST = "127.0.0.1";
static int C_PORT = 2000;
static int C_TIMEOUT = 60000;
public int stan= 0;
public void connect() throws Exception{
this.channel.setHost(C_HOST);
this.channel.setPort(C_PORT);
this.channel.setTimeout(C_TIMEOUT);
this.channel.connect();
this.mux = new ISOMUX(channel);
Thread muxThread = new Thread(mux);
muxThread.start();
System.out.println("Connected with " + channel.getHost() + ":" + channel.getPort() + " ? " + mux.isConnected());
System.out.println(channel.getSocket());
}
private void completeMsg(ISOMsg msg) {
if (msg == null) {
return;
}
Date now = new Date();
try {
msg.set(7, ISODate.getDateTime(now));
msg.set(11, String.format("%06d", stan++));
msg.set(12, ISODate.getTime(now));
msg.set(13, ISODate.getDate(now));
msg.set(15, ISODate.getDate(now));
} catch (ISOException e) {
}
}
public void signIn() throws ISOException {
ISOMsg reqMsg = new ISOMsg();
reqMsg.setPackager(packager);
reqMsg.setDirection(ISOMsg.OUTGOING);
reqMsg.setMTI("0800");
reqMsg.set(70, "001");
completeMsg(reqMsg);
if (mux.isConnected()) {
try {
ISOMsg respMsg;
respMsg = mux.request(reqMsg, C_TIMEOUT);
System.out.println("Success");
} catch (ISOException e) {
e.printStackTrace();
}
} else {
System.out.println("Error sending: not connected to server");
}
}
static int NUM_THREADS = 3;
public static void main(String[] args) {
Customizer tr = new Customizer();
try {
System.out.println("--STRESS TEST--");
System.out.println(">> CONNECTION");
tr.connect();
System.out.println(">> SIGN ON");
tr.signIn();
ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS);
List<Callable<String>> calList = new ArrayList<Callable<String>>();
long start = System.currentTimeMillis();
System.out.println(">> REQUEST WITH "+NUM_THREADS+" THREADS");
for (int i=0; i<NUM_THREADS; i++) {
// the XML_TEST is a xml string, the xml later converted to ISOMsg
Callable<String> callable = new RequestSender(tr.packager, tr.mux, ConstantData.XML_TEST, i);
calList.add(callable);
}
List<Future<String>> futList = executor.invokeAll(calList);
System.out.println(">> FINISHED");
System.out.println("elapsed ms: " + (System.currentTimeMillis() - start));
for (Future<String> fut : futList) {
try {
System.out.println(new Date()+ " | msgId: "+fut.get());
} catch (Exception e) {
e.printStackTrace();
}
}
executor.shutdown();
System.exit(1);
} catch (Exception e) {
e.printStackTrace();
}
}
}
RequestSender.java
import org.jpos.iso.ISODate;
import org.jpos.iso.ISOException;
import org.jpos.iso.ISOMUX;
import org.jpos.iso.ISOMsg;
import org.jpos.iso.packager.ISO87APackager;
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 java.io.ByteArrayInputStream;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.Callable;
import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;
public class RequestSender implements Callable<String> {
static int C_TIMEOUT = 25000; //changed
private ISO87APackager packager;
private ISOMUX mux;
private String xmlReq;
private int loop;
private int stan = 1;
public RequestSender(ISO87APackager packager, ISOMUX mux, String xmlReq, int loop) {
super();
this.packager = packager;
this.mux = mux;
this.xmlReq = xmlReq;
this.loop = loop;
}
private ISOMsg isoBuilder(String xml){
ISOMsg msg = new ISOMsg();
try {
DocumentBuilderFactory dbFactory = DocumentBuilderFactory.newInstance();
DocumentBuilder dBuilder;
dBuilder = dbFactory.newDocumentBuilder();
Document doc = dBuilder.parse(new ByteArrayInputStream(xml.getBytes()));
doc.getDocumentElement().normalize();
NodeList nList = doc.getElementsByTagName("field");
for (int temp = 0; temp < nList.getLength(); temp++) {
Node nNode = nList.item(temp);
if (nNode.getNodeType() == Node.ELEMENT_NODE) {
Element eElement = (Element) nNode;
msg.set(eElement.getAttribute("id"), eElement.getAttribute("value"));
}
}
msg.set("37", "0000000009"+loop);
completeMsg(msg);
} catch (ParserConfigurationException e) {
e.printStackTrace();
} catch (SAXException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
} catch (ISOException e) {
e.printStackTrace();
}
return msg;
}
private void completeMsg(ISOMsg msg) {
if (msg == null) {
return;
}
Date now = new Date();
try {
msg.set(7, ISODate.getDateTime(now));
msg.set(11, String.format("%06d", loop)); //changed
msg.set(12, ISODate.getTime(now));
msg.set(13, ISODate.getDate(now));
msg.set(15, ISODate.getDate(now));
} catch (ISOException e) {
}
}
private synchronized String testConfirmPayment() throws IOException {
if (mux.isConnected()) {
ISOMsg reqMsg = isoBuilder(xmlReq);
System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date())+" | REQ"+loop+"= "+reqMsg.toString().trim()+" BIT 37 : "+reqMsg.getString("37"));
try {
ISOMsg respMsg = mux.request(reqMsg, C_TIMEOUT);
if (respMsg != null){
System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date())+" | RES"+loop+"= "+respMsg.getString(48));
return respMsg.getString(48);
} else {
System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date())+" | RES"+loop+"= NO RESPONSE");
return "NO RESPONSE ";
}
} catch (ISOException e) {
e.printStackTrace();
}
}
return "NOT CONNECTED";
}
public String call() throws Exception {
return testConfirmPayment();
}
}
此请求已被服务器成功回复,但仅限于上次请求。
运行输出
--STRESS TEST--
>> CONNECTION
Connected with 127.0.0.1:2000 ? true
Socket[addr=/127.0.0.1,port=2000,localport=51086]
>> SIGN ON
Success
>> REQUEST WITH 3 THREADS
2018-01-23 09:12:06.355 | REQ0= 0200 000001 00000003 BIT 37 : 00000000090
2018-01-23 09:12:06.355 | REQ1= 0200 000001 00000003 BIT 37 : 00000000091
2018-01-23 09:12:06.355 | REQ2= 0200 000001 00000003 BIT 37 : 00000000092
2018-01-23 09:12:06.453 | RES2= 10004100108XXXXX JOHNSON SIMONE XXX
2018-01-23 09:12:11.450 | RES1= NO RESPONSE
2018-01-23 09:12:26.357 | RES0= NO RESPONSE
>> FINISHED
elapsed ms: 20045
那么如何解决呢?如果我追踪到,我会看到服务器正在响应但无法进入 MUX,尤其是在
ISOMsg respMsg = mux.request(reqMsg, C_TIMEOUT);
感谢任何回复和建议。
提前致谢
更新
问题已解决,
上面的代码现在可以正常工作了..
确保递增 STAN(序列跟踪审核编号,数据元素 11)。这就是您没有收到回复的原因,您的客户正在发送重复的消息。
我有一个可以异步处理请求的工作 JPOS ISO8583 服务器,然后我需要开发一个测试客户端来对其进行压力测试。
我想知道,多路复用器是否能够同时处理多个请求并并行处理它?请参阅此图。我正在使用 ISOMUX(不是 QMUX)。
如果答案是肯定的,那怎么办?
我试过这个代码:
Customizer.java
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import org.jpos.iso.ISODate;
import org.jpos.iso.ISOException;
import org.jpos.iso.ISOMUX;
import org.jpos.iso.ISOMsg;
import org.jpos.iso.channel.ASCIIChannel;
import org.jpos.iso.packager.ISO87APackager;
public class Customizer {
private ISO87APackager packager;
private ASCIIChannel channel;
private ISOMUX mux;
public Customizer() {
try {
this.packager = new ISO87APackager();
this.channel = new ASCIIChannel(packager);
} catch (IOException e) {
e.printStackTrace();
}
}
static String C_HOST = "127.0.0.1";
static int C_PORT = 2000;
static int C_TIMEOUT = 60000;
public int stan= 0;
public void connect() throws Exception{
this.channel.setHost(C_HOST);
this.channel.setPort(C_PORT);
this.channel.setTimeout(C_TIMEOUT);
this.channel.connect();
this.mux = new ISOMUX(channel);
Thread muxThread = new Thread(mux);
muxThread.start();
System.out.println("Connected with " + channel.getHost() + ":" + channel.getPort() + " ? " + mux.isConnected());
System.out.println(channel.getSocket());
}
private void completeMsg(ISOMsg msg) {
if (msg == null) {
return;
}
Date now = new Date();
try {
msg.set(7, ISODate.getDateTime(now));
msg.set(11, String.format("%06d", stan++));
msg.set(12, ISODate.getTime(now));
msg.set(13, ISODate.getDate(now));
msg.set(15, ISODate.getDate(now));
} catch (ISOException e) {
}
}
public void signIn() throws ISOException {
ISOMsg reqMsg = new ISOMsg();
reqMsg.setPackager(packager);
reqMsg.setDirection(ISOMsg.OUTGOING);
reqMsg.setMTI("0800");
reqMsg.set(70, "001");
completeMsg(reqMsg);
if (mux.isConnected()) {
try {
ISOMsg respMsg;
respMsg = mux.request(reqMsg, C_TIMEOUT);
System.out.println("Success");
} catch (ISOException e) {
e.printStackTrace();
}
} else {
System.out.println("Error sending: not connected to server");
}
}
static int NUM_THREADS = 3;
public static void main(String[] args) {
Customizer tr = new Customizer();
try {
System.out.println("--STRESS TEST--");
System.out.println(">> CONNECTION");
tr.connect();
System.out.println(">> SIGN ON");
tr.signIn();
ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS);
List<Callable<String>> calList = new ArrayList<Callable<String>>();
long start = System.currentTimeMillis();
System.out.println(">> REQUEST WITH "+NUM_THREADS+" THREADS");
for (int i=0; i<NUM_THREADS; i++) {
// the XML_TEST is a xml string, the xml later converted to ISOMsg
Callable<String> callable = new RequestSender(tr.packager, tr.mux, ConstantData.XML_TEST, i);
calList.add(callable);
}
List<Future<String>> futList = executor.invokeAll(calList);
System.out.println(">> FINISHED");
System.out.println("elapsed ms: " + (System.currentTimeMillis() - start));
for (Future<String> fut : futList) {
try {
System.out.println(new Date()+ " | msgId: "+fut.get());
} catch (Exception e) {
e.printStackTrace();
}
}
executor.shutdown();
System.exit(1);
} catch (Exception e) {
e.printStackTrace();
}
}
}
RequestSender.java
import org.jpos.iso.ISODate;
import org.jpos.iso.ISOException;
import org.jpos.iso.ISOMUX;
import org.jpos.iso.ISOMsg;
import org.jpos.iso.packager.ISO87APackager;
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 java.io.ByteArrayInputStream;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.Callable;
import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;
public class RequestSender implements Callable<String> {
static int C_TIMEOUT = 25000; //changed
private ISO87APackager packager;
private ISOMUX mux;
private String xmlReq;
private int loop;
private int stan = 1;
public RequestSender(ISO87APackager packager, ISOMUX mux, String xmlReq, int loop) {
super();
this.packager = packager;
this.mux = mux;
this.xmlReq = xmlReq;
this.loop = loop;
}
private ISOMsg isoBuilder(String xml){
ISOMsg msg = new ISOMsg();
try {
DocumentBuilderFactory dbFactory = DocumentBuilderFactory.newInstance();
DocumentBuilder dBuilder;
dBuilder = dbFactory.newDocumentBuilder();
Document doc = dBuilder.parse(new ByteArrayInputStream(xml.getBytes()));
doc.getDocumentElement().normalize();
NodeList nList = doc.getElementsByTagName("field");
for (int temp = 0; temp < nList.getLength(); temp++) {
Node nNode = nList.item(temp);
if (nNode.getNodeType() == Node.ELEMENT_NODE) {
Element eElement = (Element) nNode;
msg.set(eElement.getAttribute("id"), eElement.getAttribute("value"));
}
}
msg.set("37", "0000000009"+loop);
completeMsg(msg);
} catch (ParserConfigurationException e) {
e.printStackTrace();
} catch (SAXException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
} catch (ISOException e) {
e.printStackTrace();
}
return msg;
}
private void completeMsg(ISOMsg msg) {
if (msg == null) {
return;
}
Date now = new Date();
try {
msg.set(7, ISODate.getDateTime(now));
msg.set(11, String.format("%06d", loop)); //changed
msg.set(12, ISODate.getTime(now));
msg.set(13, ISODate.getDate(now));
msg.set(15, ISODate.getDate(now));
} catch (ISOException e) {
}
}
private synchronized String testConfirmPayment() throws IOException {
if (mux.isConnected()) {
ISOMsg reqMsg = isoBuilder(xmlReq);
System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date())+" | REQ"+loop+"= "+reqMsg.toString().trim()+" BIT 37 : "+reqMsg.getString("37"));
try {
ISOMsg respMsg = mux.request(reqMsg, C_TIMEOUT);
if (respMsg != null){
System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date())+" | RES"+loop+"= "+respMsg.getString(48));
return respMsg.getString(48);
} else {
System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date())+" | RES"+loop+"= NO RESPONSE");
return "NO RESPONSE ";
}
} catch (ISOException e) {
e.printStackTrace();
}
}
return "NOT CONNECTED";
}
public String call() throws Exception {
return testConfirmPayment();
}
}
此请求已被服务器成功回复,但仅限于上次请求。
运行输出
--STRESS TEST--
>> CONNECTION
Connected with 127.0.0.1:2000 ? true
Socket[addr=/127.0.0.1,port=2000,localport=51086]
>> SIGN ON
Success
>> REQUEST WITH 3 THREADS
2018-01-23 09:12:06.355 | REQ0= 0200 000001 00000003 BIT 37 : 00000000090
2018-01-23 09:12:06.355 | REQ1= 0200 000001 00000003 BIT 37 : 00000000091
2018-01-23 09:12:06.355 | REQ2= 0200 000001 00000003 BIT 37 : 00000000092
2018-01-23 09:12:06.453 | RES2= 10004100108XXXXX JOHNSON SIMONE XXX
2018-01-23 09:12:11.450 | RES1= NO RESPONSE
2018-01-23 09:12:26.357 | RES0= NO RESPONSE
>> FINISHED
elapsed ms: 20045
那么如何解决呢?如果我追踪到,我会看到服务器正在响应但无法进入 MUX,尤其是在
ISOMsg respMsg = mux.request(reqMsg, C_TIMEOUT);
感谢任何回复和建议。
提前致谢
更新
问题已解决, 上面的代码现在可以正常工作了..
确保递增 STAN(序列跟踪审核编号,数据元素 11)。这就是您没有收到回复的原因,您的客户正在发送重复的消息。