我目前正在嘗試使用純 java 將一堆簡單消息發(fā)送到隊列。public AtomicReference<Message> doSend(String message, String queue){ try (JMSContext context = connectionFactory.createContext()) { TextMessage textMessage = context.createTextMessage(message); final AtomicReference<Message> msg = new AtomicReference<>(); msg.set(textMessage); log.info("Sending message to queue {}", queue); context.createProducer().send(createDestination(context, queue), textMessage); log.info("Message sent to queue {}, messageId provided {}", queue, msg.get().getJMSMessageID()); return msg; } catch (Exception e) { log.error("Failed to send message to queue",e); throw new SipJmsException("Failed to send message to queue", e); }}private Destination createDestination(JMSContext context, String queue){ log.debug("Creating destination queue {} connection",queue); return context.createQueue(queue);}我連續(xù)發(fā)送 N 條消息，日志顯示 JMSMessageId 始終生成相同。[main] Sending message to queue TEST_QUEUE[main] Message sent to queue TEST_QUEUE, messageId provided ID:414d5120444556494d53514d20202020551c3f5d81619824[main] Sending message to queue TEST_QUEUE[main] Message sent to queue TEST_QUEUE, messageId provided ID:414d5120444556494d53514d20202020551c3f5d83619824ETC。據(jù)我所知，JMSMessageId 應(yīng)該是唯一的，它的沖突會導致問題。O'Reily 的書說：JMSMessageID 是一個唯一標識消息的字符串值。標識符的唯一性取決于供應(yīng)商。JMSMessageID 對于需要對消息進行唯一索引的 JMS 消費者應(yīng)用程序中的歷史存儲庫很有用。與 JMSCorrelationID 結(jié)合使用，JMSMessageID 也可用于關(guān)聯(lián)消息：String messageid = message.getJMSMessageID();那么，為什么 MessageId 不是唯一的呢？（應(yīng)用程序運行之間甚至相同）。
3 回答
慕斯王
TA貢獻1864條經(jīng)驗 獲得超2個贊
消息 ID 是唯一的，我用以下標記了不同的編號*：
414d5120444556494d53514d20202020551c3f5d81619824
*
414d5120444556494d53514d20202020551c3f5d83619824
撒科打諢
TA貢獻1934條經(jīng)驗 獲得超2個贊
我創(chuàng)建了一個簡單的 JMS 程序，它將 5 條消息放入一個隊列，每次放入后，它都會輸出 JMSMessageId。
示例輸出：
2019/08/13 19:15:18.824 MQTestJMS11x5: testConn: successfully connected.
2019/08/13 19:15:18.845 MQTestJMS11x5: testConn: successfully opened TEST.Q1
2019/08/13 19:15:18.845 MQTestJMS11x5: sendMsg: Sending request to queue:///TEST.Q1
2019/08/13 19:15:18.845 MQTestJMS11x5: sendMsg:
2019/08/13 19:15:18.887 MQTestJMS11x5: sendMsg: Sent message: MessageId=ID:414d51204d515754312020202020202028cd525d24201102
2019/08/13 19:15:18.887 MQTestJMS11x5: sendMsg: Sent message: MessageId=ID:414d51204d515754312020202020202028cd525d24201103
2019/08/13 19:15:18.888 MQTestJMS11x5: sendMsg: Sent message: MessageId=ID:414d51204d515754312020202020202028cd525d24201104
2019/08/13 19:15:18.889 MQTestJMS11x5: sendMsg: Sent message: MessageId=ID:414d51204d515754312020202020202028cd525d24201105
2019/08/13 19:15:18.889 MQTestJMS11x5: sendMsg: Sent message: MessageId=ID:414d51204d515754312020202020202028cd525d24201106
2019/08/13 19:15:18.892 MQTestJMS11x5: testConn: Closed session
2019/08/13 19:15:18.892 MQTestJMS11x5: testConn: Stopped connection
2019/08/13 19:15:18.893 MQTestJMS11x5: testConn: Closed connection
請注意，每個消息 ID 都是唯一的。
這是生成輸出的 JMS 程序：
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Hashtable;
import javax.jms.*;
import com.ibm.mq.jms.*;
import com.ibm.msg.client.wmq.WMQConstants;
/**
* Program Name
* MQTestJMS11x5
*
* Description
* This java JMS class will connect to a remote queue manager and put 5 messages to a queue.
*
* Sample Command Line Parameters
* -m MQA1 -h 127.0.0.1 -p 1414 -c TEST.CHL -q TEST.Q1 -u UserID -x Password
*
* @author Roger Lacroix
*/
public class MQTestJMS11x5
{
private static final SimpleDateFormat LOGGER_TIMESTAMP = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss.SSS");
private Hashtable<String,String> params;
private MQQueueConnectionFactory mqQCF = null;
/**
* The constructor
*/
public MQTestJMS11x5()
{
super();
params = new Hashtable<String,String>();
}
/**
* Make sure the required parameters are present.
* @return true/false
*/
private boolean allParamsPresent()
{
boolean b = params.containsKey("-h") && params.containsKey("-p") &&
params.containsKey("-c") && params.containsKey("-m") &&
params.containsKey("-q") &&
params.containsKey("-u") && params.containsKey("-x");
if (b)
{
try
{
Integer.parseInt((String) params.get("-p"));
}
catch (NumberFormatException e)
{
b = false;
}
}
return b;
}
/**
* Extract the command-line parameters and initialize the MQ variables.
* @param args
* @throws IllegalArgumentException
*/
private void init(String[] args) throws IllegalArgumentException
{
if (args.length > 0 && (args.length % 2) == 0)
{
for (int i = 0; i < args.length; i += 2)
{
params.put(args[i], args[i + 1]);
}
}
else
{
throw new IllegalArgumentException();
}
if (allParamsPresent())
{
try
{
mqQCF = new MQQueueConnectionFactory();
mqQCF.setQueueManager((String) params.get("-m"));
mqQCF.setHostName((String) params.get("-h"));
mqQCF.setChannel((String) params.get("-c"));
mqQCF.setTransportType(WMQConstants.WMQ_CM_CLIENT);
try
{
mqQCF.setPort(Integer.parseInt((String) params.get("-p")));
}
catch (NumberFormatException e)
{
mqQCF.setPort(1414);
}
}
catch (JMSException e)
{
MQTestJMS11x5.logger("getLinkedException()=" + e.getLinkedException());
MQTestJMS11x5.logger(e.getLocalizedMessage());
e.printStackTrace();
throw new IllegalArgumentException();
}
catch (Exception e)
{
MQTestJMS11x5.logger(e.getLocalizedMessage());
e.printStackTrace();
throw new IllegalArgumentException();
}
}
else
{
throw new IllegalArgumentException();
}
}
/**
* Test the connection to the queue manager.
* @throws MQException
*/
private void testConn()
{
QueueConnection conn = null;
QueueSession session = null;
Queue myQ = null;
try
{
conn = mqQCF.createQueueConnection((String) params.get("-u"), (String) params.get("-x"));
conn.start();
session = conn.createQueueSession(false, javax.jms.Session.AUTO_ACKNOWLEDGE);
MQTestJMS11x5.logger("successfully connected.");
myQ = session.createQueue((String) params.get("-q"));
MQTestJMS11x5.logger("successfully opened "+ (String) params.get("-q"));
MQDestination mqd = (MQDestination) myQ;
mqd.setTargetClient(WMQConstants.WMQ_CLIENT_JMS_COMPLIANT);
sendMsg( session, myQ);
}
catch (JMSException e)
{
MQTestJMS11x5.logger("getLinkedException()=" + e.getLinkedException());
MQTestJMS11x5.logger(e.getLocalizedMessage());
e.printStackTrace();
}
catch (Exception e)
{
MQTestJMS11x5.logger(e.getLocalizedMessage());
e.printStackTrace();
}
finally
{
try
{
if (session != null)
{
session.close();
MQTestJMS11x5.logger("Closed session");
}
}
catch (Exception ex)
{
MQTestJMS11x5.logger("session.close() : " + ex.getLocalizedMessage());
}
try
{
if (conn != null)
{
conn.stop();
MQTestJMS11x5.logger("Stopped connection");
}
}
catch (Exception ex)
{
MQTestJMS11x5.logger("connection.stop() : " + ex.getLocalizedMessage());
}
try
{
if (conn != null)
{
conn.close();
MQTestJMS11x5.logger("Closed connection");
}
}
catch (Exception ex)
{
MQTestJMS11x5.logger("connection.close() : " + ex.getLocalizedMessage());
}
}
}
/**
* Send a message to a queue.
* @throws MQException
*/
private void sendMsg(QueueSession session, Queue myQ) throws JMSException
{
QueueSender sender = null;
TextMessage msg = null;
try
{
MQTestJMS11x5.logger("Sending request to " + myQ.getQueueName());
MQTestJMS11x5.logger("");
sender = session.createSender(myQ);
for (int i=0; i < 5; i++)
{
msg = session.createTextMessage();
msg.setText("This is test message # " + (i+1));
sender.send(msg);
MQTestJMS11x5.logger("Sent message: MessageId="+msg.getJMSMessageID());
}
}
finally
{
try
{
if (sender != null)
sender.close();
}
catch (Exception ex)
{
MQTestJMS11x5.logger("sender.close() : " + ex.getLocalizedMessage());
}
}
}
/**
* A simple logger method
* @param data
*/
public static void logger(String data)
{
String className = Thread.currentThread().getStackTrace()[2].getClassName();
// Remove the package info.
if ( (className != null) && (className.lastIndexOf('.') != -1) )
className = className.substring(className.lastIndexOf('.')+1);
System.out.println(LOGGER_TIMESTAMP.format(new Date())+" "+className+": "+Thread.currentThread().getStackTrace()[2].getMethodName()+": "+data);
}
/**
* mainline
* @param args
*/
public static void main(String[] args)
{
MQTestJMS11x5 write = new MQTestJMS11x5();
try
{
write.init(args);
write.testConn();
}
catch (IllegalArgumentException e)
{
MQTestJMS11x5.logger("Usage: java MQTestJMS11x5 -m QueueManagerName -h host -p port -c channel -q JMS_Queue_Name -u UserID -x Password");
System.exit(1);
}
catch (Exception e)
{
MQTestJMS11x5.logger(e.getLocalizedMessage());
System.exit(1);
}
System.exit(0);
}
}
狐的傳說
TA貢獻1804條經(jīng)驗 獲得超3個贊
final AtomicReference<Message> msg = new AtomicReference<>();
你為什么使用“最終”。將其刪除并重試。
添加回答
舉報
0/150
提交
取消