我在我的项目中使用RabbitMQ。
我在我的消费者中有rabbitMQ的客户端部分的代码，并且连接需要tls1.1来连接真实的MQ。
我想在我的JUnit测试中测试这段代码，并模拟向我的消费者传递的消息。
我在google中看到几个不同工具的例子，如骆驼兔或activeMQ，但这个工具适用于amqp 1.0，而rabbitMQ只适用于amqp 0.9。
有人有这个问题吗？
谢谢！
UPDATE
这是测试从队列接收json的代码。
package com.foo.foo.queue;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.net.URL;
import java.security.*;
import java.security.cert.CertificateException;
import javax.net.ssl.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.json.JSONObject;
import com.foo.foo.Constants.Constants;
import com.foo.foo.core.ConfigurationContainer;
import com.foo.foo.policyfinders.PolicyFinder;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
public class BrokerThreadHLConsumer extends Thread {
private static BrokerThreadHLConsumer instance;
private static final Logger log = LogManager.getLogger(BrokerThreadHLConsumer.class);
private Channel channel;
private String queueName;
private PolicyFinder PolicyFinder;
private Connection connection;
private QueueingConsumer consumer;
private boolean loop;
private BrokerThreadHLConsumer() throws IOException {
ConnectionFactory factory = new ConnectionFactory();
char[] keyPassphrase = "clientrabbit".toCharArray();
KeyStore keyStoreCacerts;
ConfigurationContainer configurationContainer = ConfigurationContainer.getInstance();
String exchangeName = configurationContainer.getProperty(Constants.EXCHANGE_NAME);
String rabbitHost = configurationContainer.getProperty(Constants.RABBITMQ_SERVER_HOST_VALUE);
try {
/* Public key cacerts to connect to message queue*/
keyStoreCacerts = KeyStore.getInstance("PKCS12");
URL resourcePublicKey = this.getClass().getClassLoader().getResource("certs/client.keycert.p12");
File filePublicKey = new File(resourcePublicKey.toURI());
keyStoreCacerts.load(new FileInputStream(filePublicKey), keyPassphrase);
KeyManagerFactory keyManager;
keyManager = KeyManagerFactory.getInstance("SunX509");
keyManager.init(keyStoreCacerts, keyPassphrase);
char[] trustPassphrase = "changeit".toCharArray();
KeyStore tks;
tks = KeyStore.getInstance("JCEKS");
URL resourceCacerts = this.getClass().getClassLoader().getResource("certs/cacerts");
File fileCacerts = new File(resourceCacerts.toURI());
tks.load(new FileInputStream(fileCacerts), trustPassphrase);
TrustManagerFactory tmf;
tmf = TrustManagerFactory.getInstance("SunX509");
tmf.init(tks);
SSLContext c = SSLContext.getInstance("TLSv1.1");
c.init(keyManager.getKeyManagers(), tmf.getTrustManagers(), null);
factory.setUri(rabbitHost);
factory.useSslProtocol(c);
connection = factory.newConnection();
channel = connection.createChannel();
channel.exchangeDeclare(exchangeName, "fanout");
queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, exchangeName, "");
} catch (NoSuchAlgorithmException e) {
e.printStackTrace();
} catch (CertificateException e) {
e.printStackTrace();
} catch (KeyStoreException e) {
e.printStackTrace();
} catch (UnrecoverableKeyException e) {
e.printStackTrace();
} catch (KeyManagementException e1) {
e1.printStackTrace();
} catch (Exception e) {
log.error("Couldn't instantiate a channel with the broker installed in " + rabbitHost);
log.error(e.getStackTrace());
e.printStackTrace();
}
}
public static BrokerThreadHLConsumer getInstance() throws CertificateException, UnrecoverableKeyException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException, IOException {
if (instance == null)
instance = new BrokerThreadHLConsumer();
return instance;
}
public void run() {
if (PolicyFinder != null) {
try {
consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, consumer);
log.info("Consumer broker started and waiting for messages");
loop = true;
while (loop) {
try {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
JSONObject obj = new JSONObject(message);
log.info("Message received from broker " + obj);
if (StringUtils.isNotEmpty(message) && !PolicyFinder.managePolicySet(obj)) {
log.error("PolicySet error: error upgrading the policySet");
}
} catch (Exception e) {
log.error("Receiving message error");
log.error(e);
}
}
} catch (IOException e) {
log.error("Consumer couldn't start");
log.error(e.getStackTrace());
}
} else {
log.error("Consumer couldn't start cause of PolicyFinder is null");
}
}
public void close() {
loop = false;
try {
consumer.getChannel().basicCancel(consumer.getConsumerTag());
} catch (IOException e) {
e.printStackTrace();
}
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
}
try {
connection.close();
} catch (IOException e) {
e.printStackTrace();
}
}
public void setLuxPolicyFinder(PolicyFinder PolicyFinder) {
this.PolicyFinder = PolicyFinder;
}
}
据我了解，在这个问题中有两件事要尝试测试：
对于第一个，正在测试TLS本身，只有连接到配置了正确信任库的RabbitMQ的真实实例才能证明配置正在运行
然而，对于第二个，对于演示应用程序功能的测试（使用像Cucumber这样的工具以便于阅读），您可以尝试使用我正在处理的库：rabbitmq-mock（这就是为什么我要挖掘旧帖子）
只需将其包含为依赖项：
<dependency>
<groupId>com.github.fridujo</groupId>
<artifactId>rabbitmq-mock</artifactId>
<version>1.0.10</version>
<scope>test</scope>
</dependency>
并在单元测试中用new ConnectionFactory()替换new MockConnectionFactory()。
所以这就是我如何做到的，一些东西可能会在这里和那里隐藏必要的类实现细节，但你会得到一个提示！ :)
send()的调用是否正在发生！ public class SomeClassTest {
private Config config;
private RmqConfig rmqConfig;
private static final ObjectMapper mapper = new ObjectMapper();
private JasperServerClient jasperServerClient;
// @Mock
@InjectMocks
private RabbitMQProducer rabbitMQProducer;
private Connection phoenixConnection;
private String targetNotificationMessage;
SomeClass someClassObject;
@Before
public void setUp() {
// Mock basic stuffs
config = mock(Config.class);
Connection = mock(Connection.class);
rabbitMQProducer = mock(RabbitMQProducer.class); // Imp
jasperServerClient = mock(JasperServerClient.class);
rmqConfig = RmqConfig.builder()
.host("localhost")
.port(5672)
.userName("guest")
.password("guest")
.queueName("somequeue_name")
.prefetch(1)
.build();
final String randomMessage = "This is a waste message";
Message mockMsg = Message.forSending(randomMessage.getBytes(), null, rmqConfig.getQueueName(), rmqConfig.getQueueName(), "text/plain", "UTF-8", true); // prepare a mock message
// Prepare service configs
ConnectionConfig connectionConfig = RmqConfigUtil.getConfig(rmqConfig);
ProducerConfig producerConfig = new ProducerConfigBuilder()
.exchange(rmqConfig.getQueueName())
.contentType("text/pain")
.contentEncoding("UTF-8")
.connection(connectionConfig).build();
rabbitMQProducer.open(croducerConfig.asMap());
// build the major stuff where the code resides
someClassObject = SomeClass.builder()
.phoenixConnection(phoenixConnection)
.userExchangeName(rmqConfig.getQueueName())
.userRabbitMQProducer(rabbitMQProducer)
.ftpConfig(config.getFtpConfig())
.jasperServerClient(jasperServerClient)
.objectMapper(new ObjectMapper())
.build();
MockitoAnnotations.initMocks(this);
}
@Test
public void testNotificationPub() throws Exception {
// Prepare expected Values
targetNotificationMessage = <<some message>>
// Reflection - my target functions were private
Class cls = Class.forName("com.some.path.to.class");
Object[] objForGetMessage = {<<stuffs>>, <<stuffs>>};
Method getNotificationMessage = cls.getDeclaredMethod("private_fn_1", <<some class>>.class, <<some class>>.class);
Method pubNotification = cls.getDeclaredMethod("private_fn_2", <<some class>>.class, RabbitMQProducer.class, String.class);
getNotificationMessage.setAccessible(true);
pubNotification.setAccessible(true);
// Test Case #1
final <<some class>> notificationMessage = (<<some class>>)getNotificationMessage.invoke(someClassObject, objForGetMessage);
assertEquals(notificationMessage.getMessage(), targetNotificationMessage);
// Test Case #2 - this does RMQ call
Object[] objPubMessage = {notificationMessage, rabbitMQProducer, rmqConfig.getQueueName()};
final Object publishNotification = pubNotification.invoke(someClassObject, objPubMessage);
assertEquals(publishNotificationResp, publishNotification); //viola
//Important, since RabbitMQProducer is mocked, we need to checkup if function call is made to "send" function which send data to RMQ
verify(rabbitMQProducer,times(1)).send(any());
}
@Test
public void testMockCreation(){
assertNotNull(rmqConfig);
assertNotNull(config);
}
我知道，这是一个老问题，至今还没有答案。在同一个问题上帮助了我很多的是以下博客文章：https://tamasgyorfi.net/2016/04/21/writing-integration-tests-for-rabbitmq-based-components/。它使用Apache QPID（不是OP中建议的ActiveMQ），它支持AMQP 0.9.1。