Hi,
I've just started experimenting with the performance of the ActiveMQ CPP
client. I'm using Active MQ 4.0.2 and ActiveMQ CPP 2.0.1 on Redhat
Enterprise 3 with gcc 3.2.3. I'm comparing a simple benchmark application
written in C++ with a similar one written in Java.
I'm seeing reasonable performance (sub-millisecond latency) when sending
messages < 802 bytes, but at exactly 802 bytes the latency increases by 2
orders of magnitude. This is independent of whether I enable persistence,
async_send, change the acknowledgement mode for consumers, etc. Again, this
doesn't happen in a similar Java client. My guess is that there is a bug in
how the C++ client fragments larger messages, but it may well be a problem
with how I'm using it.
I've stripped my code down to a simple example that's very close to the
example code. Can anyone tell me what I'm doing wrong, or if this is a bug
in the ActiveMQ Cpp client?
Here's the output from 2 runs right around the breaking point:
RUN 1:
~/test/cppProducer $ ./benchmark -msgs 100 -size 802
=====================================================
Num Messages : 100
Message Length: 802 bytes
AMQ Broker URI:
tcp://localhost:61616?wireFormat=openwire&transport.useAsyncSend=true
-----------------------------------------------------
..................................................
..................................................
Sent 100 messages of size: 802
Avg Send Time = 0.0402068
Total: 4.02068
=====================================================
RUN 2:
~/test/cppProducer $ ./benchmark -msgs 100 -size 801
=====================================================
Num Messages : 100
Message Length: 801 bytes
AMQ Broker URI:
tcp://localhost:61616?wireFormat=openwire&transport.useAsyncSend=true
-----------------------------------------------------
..................................................
..................................................
Sent 100 messages of size: 801
Avg Send Time = 0.000161011
Total: 0.0161011
=====================================================
~/test/cppProducer $
Here's the sample code:
/*--- File: main.cpp --*/
#include <activemq/core/ActiveMQConnectionFactory.h>
#include <activemq/concurrent/Thread.h>
#include <activemq/concurrent/Thread.h>
#include <activemq/util/Integer.h>
#include <activemq/util/Config.h>
#include <cms/TextMessage.h>
#include <cms/BytesMessage.h>
#include <cms/MapMessage.h>
#include <cms/ExceptionListener.h>
#include <cms/MessageListener.h>
#include <stdlib.h>
#include <algorithm>
#include <numeric>
#include <iomanip>
#include <iostream>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
using namespace activemq::core;
using namespace activemq::concurrent;
using namespace cms;
using namespace std;
int main(int argc, char* argv[] ) {
string baseBrokerURI = "localhost:61616";
unsigned long numMessages = 1000;
unsigned int messageLength = 100;
std::cout << "=====================================================\n";
// Parse the command line (if any)
int nextArg = 1;
while ( nextArg < argc ) {
string nextToken = argv[ nextArg++ ];
if ( nextToken == "-msgs" ) {
numMessages = atoi( argv[ nextArg++ ]);
std::cout << "Num Messages : " << numMessages << std::endl;
} else if ( nextToken == "-size" ) {
messageLength = atoi( argv[ nextArg++ ]);
std::cout << "Message Length: " << messageLength << " bytes" <<
std::endl;
} else if ( nextToken == "-uri" ) {
baseBrokerURI = argv[ nextArg++ ];
std::cout << "Broker URI : " << baseBrokerURI << std::endl;
} else {
std::cout << "Usage:\nbenchmark [-h][-msgs count(default: 1000)][-size
messageLength(default:100)][-uri host:port]" << std::endl;
return 0;
};
};
std::string brokerURI = string( "tcp://" ) + baseBrokerURI;
brokerURI += "?wireFormat=openwire&transport.useAsyncSend=true";
std::cout << "AMQ Broker URI: " << brokerURI << std::endl;
std::cout << "-----------------------------------------------------" <<
std::endl;
// Create a ConnectionFactory
ActiveMQConnectionFactory* ConnectionFactory =
new ActiveMQConnectionFactory( brokerURI );
// Create a Connection
cms::Connection* myConnection =
ConnectionFactory->createConnection();
myConnection->start();
string myClientId = myConnection->getClientID();
// Create a Session
cms::Session* mySession =
myConnection->createSession( cms::Session::AUTO_ACKNOWLEDGE );
// Create a Destination
cms::Destination* myDestination =
mySession->createQueue( "SimpleCppTestQueue" );
// Create a Producer
cms::MessageProducer* myProducer =
mySession->createProducer( myDestination );
myProducer->setDeliveryMode( DeliveryMode::PERSISTENT );
// Create a Message to send
string text;
for ( unsigned int nextChar = 0; nextChar < messageLength; ++nextChar ) {
text += 'A' + ( nextChar % 26 );
}
TextMessage* message = mySession->createTextMessage( text );
// Start timer
struct timeval time;
gettimeofday(&time,0);
double startTime = (double)time.tv_sec + (double)time.tv_usec / 1000000;
// Send the messages
for( unsigned int i = 0; i < numMessages; ++i ) {
std::cout << ((i%50)?".":"\n.") << std::flush;
myProducer->send( message );
}
// Stop timer
gettimeofday(&time,0);
double stopTime = (double)time.tv_sec + (double)time.tv_usec / 1000000;
double elapsed = stopTime - startTime;
double avgSendTime = elapsed / numMessages;
std::cout << "\nSent " << numMessages << " messages of size: " <<
messageLength << std::endl;
std::cout << "Avg Send Time = " << avgSendTime << std::endl;
std::cout << "Total: " << elapsed << std::endl;
std::cout << "=====================================================\n";
}
Build Command:
g++ -g -pthread -I~/activemq/activemq-cpp-2.0.1-src/src/main -c -o main.o
main.cpp
g++ -o benchmark main.o -pthread
-L~/activemq/activemq-cpp-2.0.1-src/src/main -lactivemq-cpp -luuid
--
View this message in context:
http://www.nabble.com/CPP-client%3A-high-latency-for-large-messages-tf3995387s2354.html#a11346562
Sent from the ActiveMQ - User mailing list archive at Nabble.com.
