public class InventoryUpdate implements Serializable {
private static final long serialVersionUID = 1L;
private long itemId;
private int quantity;
private Date updateDate;
...
}
public class InventoryUpdateProducer { private static String url = "failover:(tcp://localhost:61616)?maxReconnectDelay=5000"; // Name of the queue we will be sending messages to private static String subject = "InventoryUpdate"; public static void main(String[] args) throws JMSException { // Getting JMS connection from the server and starting it ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url); Connection connection = connectionFactory.createConnection(); connection.start(); // JMS messages are sent and received using a Session. We will // create here a non-transactional session object. If you want // to use transactions you should set the first parameter to 'true' Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // Destination represents here our topic'InventoryUpdate'on the // JMS server. You don't have to do anything special on the // server to create it, it will be created automatically. Destination destination = session.createTopic(subject); // MessageProducer is used for sending messages (as opposed // to MessageConsumer which is used for receiving them) MessageProducer producer = session.createProducer(destination); producer.setDeliveryMode(DeliveryMode.PERSISTENT); // We will send a InventoryUpdate message. Random randomGenerator = new Random(); InventoryUpdate iUpdate = new InventoryUpdate(randomGenerator.nextInt(10), 100, new Date()); ObjectMessage objMessage = session.createObjectMessage(iUpdate); // Here we are sending the message! producer.send(objMessage); System.out.println("Sent message '" + objMessage + "'" + iUpdate); connection.close(); } }
public class InventoryUpdateConsumer {
// URL of the JMS server
private static String url = "failover:(tcp://localhost:61616)?maxReconnectDelay=5000"; // Name of the queue we will receive messages from
private static String subject = "VirtualTopicConsumers.ConsumerB.InventoryUpdate";
public static void main(String[] args) throws JMSException {
// Getting JMS connection from the server
ConnectionFactory connectionFactory
= new ActiveMQConnectionFactory(url);
Connection connection = connectionFactory.createConnection();
connection.start();
// Creating session for seding messages
Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
// Getting the queue 'InventoryUpdate'
Destination destination = session.createQueue(subject);
// MessageConsumer is used for receiving (consuming) messages
MessageConsumer consumer = session.createConsumer(destination);
// Here we receive the message.
// By default this call is blocking, which means it will wait
// for a message to arrive on the queue.
Message message = consumer.receive();
// There are many types of Message and TextMessage
// is just one of them. Producer sent us a TextMessage
// so we must cast to it to get access to its .getText()
// method.
if (message instanceof ObjectMessage){
InventoryUpdate update = (InventoryUpdate) ((ObjectMessage) message).getObject();
System.out.println(update);
}
System.out.println("Done");
connection.close();
}
}
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:camel="http://camel.apache.org/schema/spring"
xmlns:broker="http://activemq.apache.org/schema/core"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.5.0.xsd">
<!-- If a bean is defined in Spring XML or scanned using the Spring 2.5
component scanning mechanism and a <camelContext> is used or a CamelBeanPostProcessor
then we process a number of Camel annotations to do various things such as
injecting resources or producing, consuming or routing messages. -->
<!-- lets configure the Camel ActiveMQ to use the ActiveMQ broker declared
above -->
<bean id="main-broker" class="org.apache.activemq.camel.component.ActiveMQComponent">
<property name="brokerURL" value="tcp://localhost:61616" />
</bean>
<bean id="jmsToFileRoute" class="com.JmsToEmailRoute" />
<camel:camelContext id="camelContext">
<camel:routeBuilder ref="jmsToFileRoute" />
</camel:camelContext>
</beans>
import org.apache.camel.builder.RouteBuilder; import org.apache.camel.processor.idempotent.jdbc.JdbcMessageIdRepository; import com.mchange.v2.c3p0.ComboPooledDataSource; public class JmsToEmailRoute extends RouteBuilder { @Override public void configure() throws Exception { ComboPooledDataSource dataSource = new ComboPooledDataSource(); dataSource.setDriverClass("com.mysql.jdbc.Driver"); dataSource.setJdbcUrl( "jdbc:mysql://localhost:3306/test" ); dataSource.setUser("root"); dataSource.setPassword(""); from("activemq:queue:VirtualTopicConsumers.ConsumerA.InventoryUpdate") .idempotentConsumer( bean(MessageProcessor.class, "getId"), JdbcMessageIdRepository.jpaMessageIdRepository( dataSource, "notificationProcessor")) .bean(MessageProcessor.class, "process(InventoryUpdate)"); } }
import InventoryUpdate;
public class MessageProcessor {
public void process(InventoryUpdate message){
System.out.println("Got message: " + message);
}
public long getId(InventoryUpdate message){
return message.getItemId();
}
}
JmsToEmailRoute, it is defining a route from Queue
"VirtualTopicConsumers.ConsumerA.InventoryUpdate" to the processor bean
"MessageProcessor". As such, Camel will invoke "process(InventoryUpdate message)" method for every InventoryUpdate event published to the topic "InventoryUpdate".
[WebMethod]
ResponseMessage DoRequest(RequestMessage)
{
Extract the request identifier from the request.
Check if the request has already been processed by querying the persistent log.
If (request processed) then
return response
Else
{
ITransaction trans = CreateTransaction()
ProcessMessage(RequestMessage, trans);
//Store the request identifier the persistent log or database
UpdateLog(request identifier, trans)
trans.Commit();
Return responsemessage;
}
}
//Use C3P0 connection pooling
ComboPooledDataSource dataSource = new ComboPooledDataSource();
dataSource.setDriverClass("com.mysql.jdbc.Driver");
dataSource.setJdbcUrl( "jdbc:mysql://localhost:3306/test" );
dataSource.setUser("root");
dataSource.setPassword("");
from("activemq:queue:VirtualTopicConsumers.ConsumerA.InventoryUpdate")
.idempotentConsumer(
bean(MessageProcessor.class, "getId"),
JdbcMessageIdRepository.jpaMessageIdRepository(
dataSource, "notificationProcessor"))
.bean(MessageProcessor.class,
"process(InventoryUpdate)");
MessageProcessor.getId() is invoked to get the request identifier. '
JdbcMessageIdRepository.jpaMessageIdRepository ' checks this identifier against the database. If no existing identifier is found, it stores the identifier in the database and invokes
'process(InventoryUpdate)' method.iv. Finally, start the consumer.
public class TestMessageProducer {
public static void main(String[] args) {
//Talks to InventoryUpdateProducer
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("camelContext.xml");
}
}
public class EServiceImpl implements EService {
//Measure the # of times the API is invoked
private final MeterMetric eMetric = Metrics.newMeter(
getClass(), "Send API", "reqs", TimeUnit.SECONDS);
//Measure the rate of event i.e. req/ sec.
private final CounterMetric eCounter = Metrics.newCounter(getClass(),
"ECounter");
ConsoleReporter.enable(10, TimeUnit.SECONDS) //Print metrics to STDERR every 10s
public void service1(){
eCounter.inc();//Increase counter by 1
.........
eMetric.mark();//Mark the occurrence of an event.
}
7/20/11 10:41:06 AM =============================================================
com.impl.EServiceImpl:
ECounter:
count = 10
Send API:
count = 10
mean rate = 5.00 reqs/s
1-minute rate = 2.50 reqs/s
5-minute rate = 0.50 reqs/s
15-minute rate = 0.16 reqs/s
<servlet>
<servlet-name>CodahaleMetrics</servlet-name>
<servlet-class>com.yammer.metrics.reporting.MetricsServlet</servlet-class>
<init-param>
<param-name>metrics-uri</param-name>
<param-value>/metrics</param-value>
</init-param>
<init-param>
<param-name>ping-uri</param-name>
<param-value>/ping</param-value>
</init-param>
<init-param>
<param-name>healthcheck-uri</param-name>
<param-value>/health</param-value>
</init-param>
<init-param>
<param-name>threads-uri</param-name>
<param-value>/threads</param-value>
</init-param>
</servlet>
....
<servlet-mapping>
<servlet-name>CodahaleMetrics</servlet-name>
<url-pattern>/CodahaleMetrics/*</url-pattern>
</servlet-mapping>
Average Facebook metrics
| |
Click through rate
|
1%
|
Conversion rate
|
2%
|
Average friends per account
|
132
|
Average impressions per post
|
50
|
