西安网站建设优化,集团门户网站建设不足,建网站代理哪个,个人怎么做贷款网站在本文中#xff0c;我们将研究如何从Scala连接到RabbitMQ#xff0c;以便可以从应用程序中支持AMQP协议。 在此示例中#xff0c;我将使用Play Framework 2.0作为容器#xff08;有关更多信息#xff0c;请参阅我在该主题上的其他文章 #xff09;在其中运行应用程序我们将研究如何从Scala连接到RabbitMQ以便可以从应用程序中支持AMQP协议。 在此示例中我将使用Play Framework 2.0作为容器有关更多信息请参阅我在该主题上的其他文章 在其中运行应用程序因为Play使得使用Scala进行开发变得更加容易。 本文还将使用Akka actor发送和接收RabbitMQ的消息。 什么是AMQP 首先快速介绍AMQP。 AMQP代表“高级消息队列协议”是消息传递的开放标准。 AMQP 主页声明其愿景是“成为所有消息中间件之间互操作性的标准协议”。 AMQP定义了用于交换消息的传输级别协议该协议可用于集成来自许多不同平台语言和技术的应用程序。 有许多工具可以实现此协议但是RabbitMQ引起了越来越多的关注。 RabbitMQ是使用AMQP的基于Erlang的开源消息代理。 所有会说AMQP的应用程序都可以连接并使用RabbitMQ。 因此在本文中我们将展示如何将基于Play2 / Scala / Akka的应用程序连接到RabbitMQ。 在本文中我们将向您展示如何实现两种最常见的方案 发送/接收我们将配置一个发件人每隔几秒钟发送一条消息并使用两个侦听器以循环方式从队列中读取消息。 发布/订阅在此示例中我们将创建几乎相同的方案但是这次侦听器将同时获得消息。 我假设您已经安装了RabbitMQ。 如果不是请按照其网站上的说明进行操作。 设置基本的Play 2 / Scala项目 在此示例中我创建了一个新的Play 2项目。 这样做很容易 josJoss-MacBook-Pro.local:~/Dev/play-2.0-RC2$ ./play new Play2AndRabbitMQ_ _ _ __ | | __ _ _ _| |
| _ \| |/ _ | || |_|
| __/|_|\____|\__ (_)
|_| |__/ play! 2.0-RC2, http://www.playframework.orgThe new application will be created in /Users/jos/Dev/play-2.0/PlayAndRabbitMQWhat is the application name? PlayAndRabbitMQWhich template do you want to use for this new application? 1 - Create a simple Scala application2 - Create a simple Java application3 - Create an empty project 1OK, application PlayAndRabbitMQ is created.Have fun! 我曾经使用scala-ide插件在Eclipse上工作所以我执行play eclipsify并将项目导入Eclipse。 我们需要做的下一步是建立正确的依赖关系。 Play为此使用sbt并允许您从项目目录中的build.scala文件配置依赖项。 我们将添加的唯一依赖关系是RabbitMQ的Java客户端库。 即使Lift提供了一个基于Scala的AMQP库但我发现直接使用RabbitMQ也是一样容易。 添加依赖项后我的build.scala如下所示 import sbt._
import Keys._
import PlayProject._object ApplicationBuild extends Build {val appName PlayAndRabbitMQval appVersion 1.0-SNAPSHOTval appDependencies Seq(com.rabbitmq % amqp-client % 2.8.1)val main PlayProject(appName, appVersion, appDependencies, mainLang SCALA).settings()
}将rabbitMQ配置添加到配置文件 对于我们的示例我们可以配置一些东西。 将消息发送到的队列要使用的交换以及运行RabbitMQ的主机。 在实际情况下我们将需要设置更多的配置选项但是在这种情况下我们只有这三个。 将以下内容添加到您的application.conf中以便我们可以从我们的应用程序中引用它。 #rabbit-mq configuration
rabbitmq.hostlocalhost
rabbitmq.queuequeue1
rabbitmq.exchangeexchange1 现在我们可以使用ConfigFactory访问这些配置文件。 为了便于访问请创建以下对象 object Config {val RABBITMQ_HOST ConfigFactory.load().getString(rabbitmq.host);val RABBITMQ_QUEUE ConfigFactory.load().getString(rabbitmq.queue);val RABBITMQ_EXCHANGEE ConfigFactory.load().getString(rabbitmq.exchange);
}初始化与RabbitMQ的连接 在研究如何使用RabbitMQ发送和接收消息之前我们还有一个要定义的对象。 要使用RabbitMQ我们需要一个连接。 我们可以使用ConnectionFactory获得与服务器的连接。 查看javadocs以获取有关如何配置连接的更多信息。 object RabbitMQConnection {private val connection: Connection null;/*** Return a connection if one doesnt exist. Else create* a new one*/def getConnection(): Connection {connection match {case null {val factory new ConnectionFactory();factory.setHost(Config.RABBITMQ_HOST);factory.newConnection();}case _ connection}}
}在应用程序启动时启动侦听器 在查看RabbitMQ代码之前我们还需要做一件事。 我们需要确保在应用程序启动时注册了消息侦听器并且发件人开始发送。 播放2提供了 为此的GlobalSettings对象您可以在应用程序启动时扩展该对象以执行代码。 对于我们的示例我们将使用以下对象请记住该对象需要存储在默认名称空间中 import play.api.mvc._
import play.api._
import rabbitmq.Senderobject Global extends GlobalSettings {override def onStart(app: Application) {Sender.startSending}
} 我们将在下面的部分中查看此Sender.startSending操作该操作将初始化所有发送者和接收者。 设置发送和接收方案 让我们看一下Sender.startSending代码该代码将设置一个将msg发送到特定队列的发送方。 为此我们使用以下代码 object Sender {def startSending {// create the connectionval connection RabbitMQConnection.getConnection();// create the channel we use to sendval sendingChannel connection.createChannel();// make sure the queue exists we want to send tosendingChannel.queueDeclare(Config.RABBITMQ_QUEUE, false, false, false, null);Akka.system.scheduler.schedule(2 seconds, 1 seconds, Akka.system.actorOf(Props(new SendingActor(channel sendingChannel, queue Config.RABBITMQ_QUEUE))), MSG to Queue);}
}class SendingActor(channel: Channel, queue: String) extends Actor {def receive {case some: String {val msg (some : System.currentTimeMillis());channel.basicPublish(, queue, null, msg.getBytes());Logger.info(msg);}case _ {}}
} 在此代码中我们采取以下步骤 使用工厂检索到RabbitMQ的连接 在此连接上创建一个通道用于与RabbitMQ通信 使用通道创建队列如果尚不存在 安排Akka每秒向演员发送一条消息。 所有这些都应该非常简单。 唯一有点复杂的部分是调度部分。 此调度操作的作用是这样的。 我们告诉Akka安排要发送给演员的消息。 我们希望在发射之前有2秒的延迟并且我们希望每秒重复一次此作业。 应该用于此的actor是SendingActor您也可以在此清单中看到。 该参与者需要访问通道以发送消息并且该参与者还需要知道接收消息的位置。 这是队列。 因此该Actor每秒将收到一条消息附加一个时间戳并使用提供的通道将此消息发送到队列channel.basicPublish“”queuenullmsg.getBytes;。 现在我们每秒发送一条消息在此队列上有可以接收消息的侦听器将是很好的。 为了接收消息我们还创建了一个Actor可以在特定队列上无限期地进行监听。 class ListeningActor(channel: Channel, queue: String, f: (String) Any) extends Actor {// called on the initial rundef receive {case _ startReceving}def startReceving {val consumer new QueueingConsumer(channel);channel.basicConsume(queue, true, consumer);while (true) {// wait for the messageval delivery consumer.nextDelivery();val msg new String(delivery.getBody());// send the message to the provided callback function// and execute this in a subactorcontext.actorOf(Props(new Actor {def receive {case some: String f(some);}})) ! msg}}
} 这个actor比我们以前发送的actor要复杂一些。 当该参与者收到消息消息的种类无关紧要时它将开始侦听创建该消息的队列。 它通过使用提供的通道创建使用者来实现此目的并告诉使用者开始在指定队列上侦听。 Consumer.nextDelivery方法将阻塞直到消息在配置的队列中等待。 收到消息后将创建一个新的Actor将消息发送到该Actor。 这个新的参与者将消息传递给提供的方法您可以在其中放置业务逻辑。 要使用此侦听器我们需要提供以下参数 频道允许访问RabbitMQ 队列监听消息的队列 f收到消息后我们将执行的功能。 第一个示例的最后一步是将所有内容粘合在一起。 为此我们向Sender.startSending方法添加了几个方法调用。 def startSending {...val callback1 (x: String) Logger.info(Recieved on queue callback 1: x);setupListener(connection.createChannel(),Config.RABBITMQ_QUEUE, callback1);// create an actor that starts listening on the specified queue and passes the// received message to the provided callbackval callback2 (x: String) Logger.info(Recieved on queue callback 2: x);// setup the listener that sends to a specific queue using the SendingActorsetupListener(connection.createChannel(),Config.RABBITMQ_QUEUE, callback2);...}private def setupListener(receivingChannel: Channel, queue: String, f: (String) Any) {Akka.system.scheduler.scheduleOnce(2 seconds, Akka.system.actorOf(Props(new ListeningActor(receivingChannel, queue, f))), );} 在此代码中您可以看到我们定义了一个回调函数并使用此回调函数以及队列和通道来创建ListeningActor。 我们使用scheduleOnce方法在单独的线程中启动此侦听器。 现在使用此代码我们可以运行应用程序播放运行打开localhost9000来启动应用程序我们应该看到类似以下输出的内容。 [info] play - Starting application default Akka system.
[info] play - Application started (Dev)
[info] application - MSG to Exchange : 1334324531424
[info] application - MSG to Queue : 1334324531424
[info] application - Recieved on queue callback 2: MSG to Queue : 1334324531424
[info] application - MSG to Exchange : 1334324532522
[info] application - MSG to Queue : 1334324532522
[info] application - Recieved on queue callback 1: MSG to Queue : 1334324532522
[info] application - MSG to Exchange : 1334324533622
[info] application - MSG to Queue : 1334324533622
[info] application - Recieved on queue callback 2: MSG to Queue : 1334324533622
[info] application - MSG to Exchange : 1334324534722
[info] application - MSG to Queue : 1334324534722
[info] application - Recieved on queue callback 1: MSG to Queue : 1334324534722
[info] application - MSG to Exchange : 1334324535822
[info] application - MSG to Queue : 1334324535822
[info] application - Recieved on queue callback 2: MSG to Queue : 1334324535822 在这里您可以清楚地看到消息的循环处理方式。 设置发布和订阅方案 一旦我们运行了上面的代码添加发布/订阅功能就变得非常简单。 现在我们使用PublishingActor代替SendingActor class PublishingActor(channel: Channel, exchange: String) extends Actor {/*** When we receive a message we sent it using the configured channel*/def receive {case some: String {val msg (some : System.currentTimeMillis());channel.basicPublish(exchange, , null, msg.getBytes());Logger.info(msg);}case _ {}}
} RabbitMQ使用交换来允许多个收件人接收相同的消息以及许多其他高级功能。 来自其他参与者的代码唯一的变化是这次我们将消息发送到交换而不是队列。 侦听器代码完全相同我们唯一需要做的就是将队列连接到特定的交换机。 这样该队列上的侦听器就可以接收发送到交换机的消息。 我们再次根据之前使用的设置方法执行此操作。 ...// create a new sending channel on which we declare the exchangeval sendingChannel2 connection.createChannel();sendingChannel2.exchangeDeclare(Config.RABBITMQ_EXCHANGEE, fanout);// define the two callbacks for our listenersval callback3 (x: String) Logger.info(Recieved on exchange callback 3: x);val callback4 (x: String) Logger.info(Recieved on exchange callback 4: x);// create a channel for the listener and setup the first listenerval listenChannel1 connection.createChannel();setupListener(listenChannel1,listenChannel1.queueDeclare().getQueue(), Config.RABBITMQ_EXCHANGEE, callback3);// create another channel for a listener and setup the second listenerval listenChannel2 connection.createChannel();setupListener(listenChannel2,listenChannel2.queueDeclare().getQueue(), Config.RABBITMQ_EXCHANGEE, callback4);// create an actor that is invoked every two seconds after a delay of// two seconds with the message msgAkka.system.scheduler.schedule(2 seconds, 1 seconds, Akka.system.actorOf(Props(new PublishingActor(channel sendingChannel2, exchange Config.RABBITMQ_EXCHANGEE))), MSG to Exchange);... 我们还为setupListener创建了一个重载方法该方法作为一个附加参数也接受要使用的交换的名称。 private def setupListener(channel: Channel, queueName : String, exchange: String, f: (String) Any) {channel.queueBind(queueName, exchange, );Akka.system.scheduler.scheduleOnce(2 seconds, Akka.system.actorOf(Props(new ListeningActor(channel, queueName, f))), );} 在这小段代码中您可以看到我们将提供的队列在我们的示例中是一个随机名称绑定到指定的交易所。 之后我们将创建一个新的监听器如我们之前所见。 现在运行此代码将产生以下输出 [info] play - Application started (Dev)
[info] application - MSG to Exchange : 1334325448907
[info] application - MSG to Queue : 1334325448907
[info] application - Recieved on exchange callback 3: MSG to Exchange : 1334325448907
[info] application - Recieved on exchange callback 4: MSG to Exchange : 1334325448907
[info] application - MSG to Exchange : 1334325450006
[info] application - MSG to Queue : 1334325450006
[info] application - Recieved on exchange callback 4: MSG to Exchange : 1334325450006
[info] application - Recieved on exchange callback 3: MSG to Exchange : 1334325450006 如您所见在这种情况下两个侦听器都收到相同的消息。 这几乎涵盖了本文的全部内容。 如您所见为RabbitMQ使用基于Java的客户端api绰绰有余并且可以从Scala轻松使用。 请注意尽管该示例尚未准备好投入生产但您应注意关闭连接并很好地关闭侦听器和参与者。 这里没有显示所有关闭代码。 参考通过我们的JCG合作伙伴 Jos Dirksen在Smart Java博客上使用ScalaPlay和Akka连接到RabbitMQAMQP 。 翻译自: https://www.javacodegeeks.com/2012/04/connect-to-rabbitmq-amqp-using-scala.html
