zoukankan      html  css  js  c++  java
  • SSM商城项目(八)

     1.   学习计划

    1、solr集群搭建

    2、使用solrj管理solr集群

    3、把搜索功能切换到集群版

    4、添加商品同步到索引库

    2.   什么是SolrCloud

    SolrCloud(solr 云)是Solr提供的分布式搜索方案,当你需要大规模,容错,分布式索引和检索能力时使用 SolrCloud。当一个系统的索引数据量少的时候是不需要使用SolrCloud的,当索引量很大,搜索请求并发很高,这时需要使用SolrCloud来满足这些需求。

     SolrCloud是基于Solr和Zookeeper的分布式搜索方案,它的主要思想是使用Zookeeper作为集群的配置信息中心。

    它有几个特色功能:

    1)集中式的配置信息

    2)自动容错

    3)近实时搜索

    4)查询时自动负载均衡

    3.   Solr集群的系统架构

     

    3.1. 物理结构

    三个Solr实例( 每个实例包括两个Core),组成一个SolrCloud。

    3.2. 逻辑结构

    索引集合包括两个Shard(shard1和shard2),shard1和shard2分别由三个Core组成,其中一个Leader两个Replication,Leader是由zookeeper选举产生,zookeeper控制每个shard上三个Core的索引数据一致,解决高可用问题。

    用户发起索引请求分别从shard1和shard2上获取,解决高并发问题。

    3.2.1.    collection

    Collection在SolrCloud集群中是一个逻辑意义上的完整的索引结构。它常常被划分为一个或多个Shard(分片),它们使用相同的配置信息。

    比如:针对商品信息搜索可以创建一个collection。

     collection=shard1+shard2+....+shardX

    3.2.2.    Core

    每个Core是Solr中一个独立运行单位,提供 索引和搜索服务。一个shard需要由一个Core或多个Core组成。由于collection由多个shard组成所以collection一般由多个core组成。

    3.2.3.    Master或Slave

    Master是master-slave结构中的主结点(通常说主服务器),Slave是master-slave结构中的从结点(通常说从服务器或备服务器)。同一个Shard下master和slave存储的数据是一致的,这是为了达到高可用目的。

    3.2.4.    Shard

    Collection的逻辑分片。每个Shard被化成一个或者多个replication,通过选举确定哪个是Leader。

    3.3. 需要实现的solr集群架构

    Zookeeper作为集群的管理工具。

    1、集群管理:容错、负载均衡。

    2、配置文件的集中管理

    3、集群的入口

    需要实现zookeeper 高可用。需要搭建集群。建议是奇数节点。需要三个zookeeper服务器。 

    搭建solr集群需要7台服务器。

    搭建伪分布式:

    需要三个zookeeper节点

    需要四个tomcat节点。

    建议虚拟机的内容1G以上。

    4.   环境准备

        CentOS-6.5-i386-bin-DVD1.iso

           jdk-7u72-linux-i586.tar.gz

        apache-tomcat-7.0.47.tar.gz

        zookeeper-3.4.6.tar.gz

        solr-4.10.3.tgz

    5.   安装步骤

    5.1. Zookeeper集群搭建

    第一步:需要安装jdk环境。

    第二步:把zookeeper的压缩包上传到服务器。

    第三步:解压缩。

    第四步:把zookeeper复制三份。

    [root@localhost ~]# mkdir /usr/local/solr-cloud

    [root@localhost ~]# cp -r zookeeper-3.4.6 /usr/local/solr-cloud/zookeeper01

    [root@localhost ~]# cp -r zookeeper-3.4.6 /usr/local/solr-cloud/zookeeper02

    [root@localhost ~]# cp -r zookeeper-3.4.6 /usr/local/solr-cloud/zookeeper03

    第五步:在每个zookeeper目录下创建一个data目录。

    第六步:在data目录下创建一个myid文件,文件名就叫做“myid”。内容就是每个实例的id。例如1、2、3

    [root@localhost data]# touch  myid

    [root@localhost data]# vim myid

    [root@localhost data]# cat myid

    1

    第七步:修改配置文件。把conf目录下的zoo_sample.cfg文件改名为zoo.cfg

     第八步:启动每个zookeeper实例。

    [root@localhost solr-cloud]# vim starall.sh
    cd zookeeper01/bin
    ./zkServer.sh start
    cd ../../
    cd zookeeper02/bin
    ./zkServer.sh start
    cd ../../
    cd zookeeper03/bin
    ./zkServer.sh start
    cd ../../

     [root@localhost solr-cloud]# chmod u+x starall.sh

    [root@localhost solr-cloud]# ./starall.sh

     

    查看zookeeper的状态:

    zookeeper01/bin/zkServer.sh status

    zookeeper02/bin/zkServer.sh status

    zookeeper03/bin/zkServer.sh status

    如果还是单机模式启动(Mode: standalone),关闭重启下。

    5.2. Solr集群的搭建

    第一步:在 /usr/local/solr-cloud创建四个tomcat实例。每个tomcat运行在不同的端口。8180、8280、8380、8480

    第二步:部署solr的war包。把单机版的solr工程复制到集群中的tomcat中。

    [root@localhost solr-cloud]# cp -r ../solr/tomcat/webapps/solr/  tomcat01/webapps/

    [root@localhost solr-cloud]# cp -r ../solr/tomcat/webapps/solr/  tomcat02/webapps/

    [root@localhost solr-cloud]# cp -r ../solr/tomcat/webapps/solr/  tomcat03/webapps/

    [root@localhost solr-cloud]# cp -r ../solr/tomcat/webapps/solr/  tomcat04/webapps/

    第三步:为每个solr实例创建一个对应的solrhome。使用单机版的solrhome复制四份。

    [root@localhost solr-cloud]# cp -r ../solr/solrhome/ solrhome01

    [root@localhost solr-cloud]# cp -r ../solr/solrhome/ solrhome02

    [root@localhost solr-cloud]# cp -r ../solr/solrhome/ solrhome03

    [root@localhost solr-cloud]# cp -r ../solr/solrhome/ solrhome04

    第四步:需要修改solr的web.xml文件。把solrhome关联起来。

    [root@localhost solr-cloud]# vim tomcat01/webapps/solr/WEB-INF/web.xml

    [root@localhost solr-cloud]# vim tomcat02/webapps/solr/WEB-INF/web.xml

    [root@localhost solr-cloud]# vim tomcat03/webapps/solr/WEB-INF/web.xml

    [root@localhost solr-cloud]# vim tomcat04/webapps/solr/WEB-INF/web.xml

    第五步:配置solrCloud相关的配置。每个solrhome下都有一个solr.xml,把其中的ip及端口号配置好。

    第六步:让zookeeper统一管理配置文件。需要把solrhome/collection1/conf目录上传到zookeeper。上传任意solrhome中的配置文件即可。

    使用工具上传配置文件:/heima/solr-4.10.3/example/scripts/cloud-scripts

    [root@localhost cloud-scripts]# ./zkcli.sh -zkhost 192.168.25.128:2181,192.168.25.128:2182,192.168.25.128:2183 -cmd upconfig -confdir /usr/local/solr-cloud/solrhome01/collection1/conf -confname myconf

    [root@localhost cloud-scripts]# cd /usr/local/solr-cloud/zookeeper01/bin

    [root@localhost bin]# ./zkCli.sh -server 192.168.25.128:2183

    第七步:修改tomcat/bin目录下的catalina.sh 文件,关联solr和zookeeper。

    把此配置添加到配置文件中:

    [root@localhost solr-cloud]# vim tomcat01/bin/catalina.sh

    JAVA_OPTS="-DzkHost=192.168.25.128:2181,192.168.25.128:2182,192.168.25.128:2183"

    [root@localhost solr-cloud]# vim tomcat02/bin/catalina.sh

    JAVA_OPTS="-DzkHost=192.168.25.128:2181,192.168.25.128:2182,192.168.25.128:2183"

    [root@localhost solr-cloud]# vim tomcat03/bin/catalina.sh

    JAVA_OPTS="-DzkHost=192.168.25.128:2181,192.168.25.128:2182,192.168.25.128:2183"

    [root@localhost solr-cloud]# vim tomcat04/bin/catalina.sh

    JAVA_OPTS="-DzkHost=192.168.25.128:2181,192.168.25.128:2182,192.168.25.128:2183"

    第八步:启动每个tomcat实例。要包装zookeeper集群是启动状态。

    [root@localhost solr-cloud]# vim start-tomcat-all.sh

    /usr/local/solr-cloud/tomcat01/bin/startup.sh
    /usr/local/solr-cloud/tomcat02/bin/startup.sh
    /usr/local/solr-cloud/tomcat03/bin/startup.sh
    /usr/local/solr-cloud/tomcat04/bin/startup.sh

    [root@localhost solr-cloud]# chmod u+x start-tomcat-all.sh
    [root@localhost solr-cloud]# ./start-tomcat-all.sh

    第九步:访问集群

    第十步:创建新的Collection进行分片处理。

    http://192.168.25.128:8180/solr/admin/collections?action=CREATE&name=collection2&numShards=2&replicationFactor=2

    第十一步:删除不用的Collection。

    http://192.168.25.128:8180/solr/admin/collections?action=DELETE&name=collection1

     

    6.   使用solrJ管理集群

    6.1. 添加文档

    使用步骤:

    第一步:把solrJ相关的jar包添加到工程中。

    第二步:创建一个SolrServer对象,需要使用CloudSolrServer子类。构造方法的参数是zookeeper的地址列表。

    第三步:需要设置DefaultCollection属性。

    第四步:创建一SolrInputDocument对象。

    第五步:向文档对象中添加域

    第六步:把文档对象写入索引库。

    第七步:提交。

        @Test
        public void testSolrCloudAddDocument() throws Exception {
            // 第一步:把solrJ相关的jar包添加到工程中。
            // 第二步:创建一个SolrServer对象,需要使用CloudSolrServer子类。构造方法的参数是zookeeper的地址列表。
            //参数是zookeeper的地址列表,使用逗号分隔
            CloudSolrServer solrServer = new CloudSolrServer("192.168.25.128:2181,192.168.25.128:2182,192.168.25.128:2183");
            // 第三步:需要设置DefaultCollection属性。
            solrServer.setDefaultCollection("collection2");
            // 第四步:创建一SolrInputDocument对象。
            SolrInputDocument document = new SolrInputDocument();
            // 第五步:向文档对象中添加域
            document.addField("item_title", "测试商品");
            document.addField("item_price", "100");
            document.addField("id", "test001");
            // 第六步:把文档对象写入索引库。
            solrServer.add(document);
            // 第七步:提交。
            solrServer.commit();
    
        }

    6.2. 查询文档

        @Test
        public void testQueryDocument() throws Exception {
            //创建一个CloudSolrServer对象
            CloudSolrServer cloudSolrServer = new CloudSolrServer("192.168.25.128:2181,192.168.25.128:2182,192.168.25.128:2183");
            //设置默认的Collection
            cloudSolrServer.setDefaultCollection("collection2");
            //创建一个查询对象
            SolrQuery query = new SolrQuery();
            //设置查询条件
            query.setQuery("*:*");
            //执行查询
            QueryResponse queryResponse = cloudSolrServer.query(query);
            //取查询结果
            SolrDocumentList solrDocumentList = queryResponse.getResults();
            System.out.println("总记录数:" + solrDocumentList.getNumFound());
            //打印
            for (SolrDocument solrDocument : solrDocumentList) {
                System.out.println(solrDocument.get("id"));
                System.out.println(solrDocument.get("title"));
                System.out.println(solrDocument.get("item_title"));
                System.out.println(solrDocument.get("item_price"));
            }
        }

    7.   把搜索功能切换到集群版

        <!-- 单机版solr服务配置 -->
    <!--     <bean id="httpSolrServer" class="org.apache.solr.client.solrj.impl.HttpSolrServer">
            <constructor-arg index="0" value="http://192.168.25.128:8090/solr/collection1"/>
        </bean> -->
        
        <!-- 集群版solr服务 -->
        <bean id="cloudSolrServer" class="org.apache.solr.client.solrj.impl.CloudSolrServer">
            <constructor-arg name="zkHost" value="192.168.25.128:2181,192.168.25.128:2182,192.168.25.128:2183"></constructor-arg>    
            <property name="defaultCollection" value="collection2"></property>
        </bean>

    启动注册中心

    [root@localhost //]# cd heima/zookeeper-3.4.6/bin

    [root@localhost bin]# ./zkServer.sh start

    启动redis集群

    [root@localhost //]# cd  usr/local/redis-cluster

    [root@localhost redis-cluster]# ./start-all.sh 

    8.   全局异常处理

    8.1. 处理思路

    8.2. 创建全局异常处理器

    package cn.e3mall.search.exception;
    
    import javax.servlet.http.HttpServletRequest;
    import javax.servlet.http.HttpServletResponse;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.web.servlet.HandlerExceptionResolver;
    import org.springframework.web.servlet.ModelAndView;
    
    public class GlobalExceptionReslover implements HandlerExceptionResolver{
        
        Logger logger = LoggerFactory.getLogger(GlobalExceptionReslover.class);
        
        @Override
        public ModelAndView resolveException(HttpServletRequest request, HttpServletResponse response, Object handler,
                Exception ex) {
            //写日志文件
            logger.error("系统发生异常", ex);
            //发邮件、发短信
            //Jmail:可以查找相关的资料
            //需要在购买短信。调用第三方接口即可。
            //展示错误页面
            ModelAndView modelAndView = new ModelAndView();
            modelAndView.setViewName("error/exception");
            return modelAndView;
        }
    }

    8.3. 上传日志文件

    log4j.rootLogger=DEBUG,A3,STDOUT
    
    log4j.appender.STDOUT=org.apache.log4j.ConsoleAppender
    log4j.appender.STDOUT.layout=org.apache.log4j.PatternLayout
    log4j.appender.STDOUT.layout.ConversionPattern=[%p] [%l] %10.10c - %m%n
    
    log4j.appender.A3=org.apache.log4j.RollingFileAppender
    log4j.appender.A3.file=logs/server.log
    log4j.appender.A3.MaxFileSize=1024KB
    log4j.appender.A3.MaxBackupIndex=10
    log4j.appender.A3.layout=org.apache.log4j.PatternLayout
    log4j.appender.A3.layout.ConversionPattern=
    
    [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n

    8.4. Springmvc中配置异常处理器

    <!-- 全局异常处理器 -->
    <bean class="cn.e3mall.search.exception.GlobalExceptionReslover"/>

    9.   同步索引库分析

    方案一:在taotao-manager中,添加商品的业务逻辑中,添加一个同步索引库的业务逻辑。

    缺点:业务逻辑耦合度高,业务拆分不明确

    方案二:业务逻辑在taotao-search中实现,调用服务在taotao-manager实现。业务逻辑分开。

    缺点:服务之间的耦合度变高。服务的启动有先后顺序。

    方案三:使用消息队列。MQ是一个消息中间件。

    MQ是一个消息中间件,ActiveMQRabbitMQkafka

    10.   ActiveMQ

    10.1. 什么是ActiveMQ

    ActiveMQ 是Apache出品,最流行的,能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,尽管JMS规范出台已经是很久的事情了,但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。

    主要特点:

    1. 多种语言和协议编写客户端。语言: Java, C, C++, C#, Ruby, Perl, Python, PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP

    2. 完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)

    3. 对Spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性

    4. 通过了常见J2EE服务器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的测试,其中通过JCA 1.5 resource adaptors的配置,可以让ActiveMQ可以自动的部署到任何兼容J2EE 1.4 商业服务器上

    5. 支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA

    6. 支持通过JDBC和journal提供高速的消息持久化

    7. 从设计上保证了高性能的集群,客户端-服务器,点对点

    8. 支持Ajax

    9. 支持与Axis的整合

    10. 可以很容易得调用内嵌JMS provider,进行测试

    10.2. ActiveMQ的消息形式

    对于消息的传递有两种类型:

    一种是点对点的,即一个生产者和一个消费者一一对应;

    另一种是发布/订阅模式,即一个生产者产生消息并进行发送后,可以由多个消费者进行接收。

    JMS定义了五种不同的消息正文格式,以及调用的消息类型,允许你发送并接收以一些不同形式的数据,提供现有消息格式的一些级别的兼容性。

      · StreamMessage -- Java原始值的数据流

      · MapMessage--一套名称-值对

      · TextMessage--一个字符串对象

      · ObjectMessage--一个序列化的 Java对象

      · BytesMessage--一个字节的数据流

    11.   ActiveMQ的安装

    进入http://activemq.apache.org/下载ActiveMQ

    11.1. 安装环境:

    1、需要jdk

    2、安装Linux系统。生产环境都是Linux系统。

    11.2. 安装步骤

    第一步: 把ActiveMQ 的压缩包上传到Linux系统。

    第二步:解压缩。

    第三步:启动。

    使用bin目录下的activemq命令启动:

    [root@localhost bin]# ./activemq start

    关闭:

    [root@localhost bin]# ./activemq stop

    查看状态:

    [root@localhost bin]# ./activemq status

    注意:如果ActiveMQ整合spring使用不要使用activemq-all-5.12.0.jar包。建议使用5.11.2

    进入管理后台:

    http://192.168.25.168:8161/admin

    用户名:admin

    密码:admin

    12.   ActiveMQ的使用方法

    12.1. Queue

    12.1.1.    Producer

    生产者:生产消息,发送端。

    把jar包添加到工程e3-manager-service的pom.xml中。使用5.11.2版本的jar包。

        <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-all</artifactId>
            </dependency>

    第一步:创建ConnectionFactory对象,需要指定服务端ip及端口号。

    第二步:使用ConnectionFactory对象创建一个Connection对象。

    第三步:开启连接,调用Connection对象的start方法。

    第四步:使用Connection对象创建一个Session对象。

    第五步:使用Session对象创建一个Destination对象(topic、queue),此处创建一个Queue对象。

    第六步:使用Session对象创建一个Producer对象。

    第七步:创建一个Message对象,创建一个TextMessage对象。

    第八步:使用Producer对象发送消息。

    第九步:关闭资源。

    package cn.e3mall.activemq;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.MessageProducer;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    
    
    public class ActiveMqTest {
        
        @Test
        public void testQueueProducer() throws Exception {
            // 第一步:创建ConnectionFactory对象,需要指定服务端ip及端口号。
            //brokerURL服务器的ip及端口号
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
            // 第二步:使用ConnectionFactory对象创建一个Connection对象。
            Connection connection = connectionFactory.createConnection();
            // 第三步:开启连接,调用Connection对象的start方法。
            connection.start();
            // 第四步:使用Connection对象创建一个Session对象。
            //第一个参数:是否开启事务。true:开启事务,第二个参数忽略。
            //第二个参数:当第一个参数为false时,才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 第五步:使用Session对象创建一个Destination对象(topic、queue),此处创建一个Queue对象。
            //参数:队列的名称。
            Queue queue = session.createQueue("test-queue");
            // 第六步:使用Session对象创建一个Producer对象。
            MessageProducer producer = session.createProducer(queue);
            // 第七步:创建一个Message对象,创建一个TextMessage对象。
            /*TextMessage message = new ActiveMQTextMessage();
            message.setText("hello activeMq,this is my first test.");*/
            TextMessage textMessage = session.createTextMessage("hello activeMq,this is my first test.");
            // 第八步:使用Producer对象发送消息。
            producer.send(textMessage);
            // 第九步:关闭资源。
            producer.close();
            session.close();
            connection.close();
        }
    
    }

    12.1.2.    Consumer

    消费者:接收消息。

    第一步:创建一个ConnectionFactory对象。

    第二步:从ConnectionFactory对象中获得一个Connection对象。

    第三步:开启连接。调用Connection对象的start方法。

    第四步:使用Connection对象创建一个Session对象。

    第五步:使用Session对象创建一个Destination对象。和发送端保持一致queue,并且队列的名称一致。

    第六步:使用Session对象创建一个Consumer对象。

    第七步:接收消息。

    第八步:打印消息。

    第九步:关闭资源

        @Test
        public void testQueueConsumer() throws Exception {
            // 第一步:创建一个ConnectionFactory对象。
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
            // 第二步:从ConnectionFactory对象中获得一个Connection对象。
            Connection connection = connectionFactory.createConnection();
            // 第三步:开启连接。调用Connection对象的start方法。
            connection.start();
            // 第四步:使用Connection对象创建一个Session对象。
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 第五步:使用Session对象创建一个Destination对象。和发送端保持一致queue,并且队列的名称一致。
            Queue queue = session.createQueue("test-queue");
            // 第六步:使用Session对象创建一个Consumer对象。
            MessageConsumer consumer = session.createConsumer(queue);
            // 第七步:接收消息。
            consumer.setMessageListener(new MessageListener() {
                
                @Override
                public void onMessage(Message message) {
                    try {
                        TextMessage textMessage = (TextMessage) message;
                        String text = null;
                        //取消息的内容
                        text = textMessage.getText();
                        // 第八步:打印消息。
                        System.out.println(text);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            //等待键盘输入
            System.in.read();
            // 第九步:关闭资源
            consumer.close();
            session.close();
            connection.close();
        }

    12.2. Topic

    12.2.1.    Producer

    使用步骤:

    第一步:创建ConnectionFactory对象,需要指定服务端ip及端口号。

    第二步:使用ConnectionFactory对象创建一个Connection对象。

    第三步:开启连接,调用Connection对象的start方法。

    第四步:使用Connection对象创建一个Session对象。

    第五步:使用Session对象创建一个Destination对象(topic、queue),此处创建一个Topic对象。

    第六步:使用Session对象创建一个Producer对象。

    第七步:创建一个Message对象,创建一个TextMessage对象。

    第八步:使用Producer对象发送消息。

    第九步:关闭资源。

    @Test
        public void testTopicProducer() throws Exception {
            // 第一步:创建ConnectionFactory对象,需要指定服务端ip及端口号。
            // brokerURL服务器的ip及端口号
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
            // 第二步:使用ConnectionFactory对象创建一个Connection对象。
            Connection connection = connectionFactory.createConnection();
            // 第三步:开启连接,调用Connection对象的start方法。
            connection.start();
            // 第四步:使用Connection对象创建一个Session对象。
            // 第一个参数:是否开启事务。true:开启事务,第二个参数忽略。
            // 第二个参数:当第一个参数为false时,才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 第五步:使用Session对象创建一个Destination对象(topic、queue),此处创建一个topic对象。
            // 参数:话题的名称。
            Topic topic = session.createTopic("test-topic");
            // 第六步:使用Session对象创建一个Producer对象。
            MessageProducer producer = session.createProducer(topic);
            // 第七步:创建一个Message对象,创建一个TextMessage对象。
            /*
             * TextMessage message = new ActiveMQTextMessage(); message.setText(
             * "hello activeMq,this is my first test.");
             */
            TextMessage textMessage = session.createTextMessage("hello activeMq,this is my topic test");
            // 第八步:使用Producer对象发送消息。
            producer.send(textMessage);
            // 第九步:关闭资源。
            producer.close();
            session.close();
            connection.close();
        }

    12.2.2.    Consumer

    消费者:接收消息。

    第一步:创建一个ConnectionFactory对象。

    第二步:从ConnectionFactory对象中获得一个Connection对象。

    第三步:开启连接。调用Connection对象的start方法。

    第四步:使用Connection对象创建一个Session对象。

    第五步:使用Session对象创建一个Destination对象。和发送端保持一致topic,并且话题的名称一致。

    第六步:使用Session对象创建一个Consumer对象。

    第七步:接收消息。

    第八步:打印消息。

    第九步:关闭资源

    @Test
        public void testTopicConsumer() throws Exception {
            // 第一步:创建一个ConnectionFactory对象。
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
            // 第二步:从ConnectionFactory对象中获得一个Connection对象。
            Connection connection = connectionFactory.createConnection();
            // 第三步:开启连接。调用Connection对象的start方法。
            connection.start();
            // 第四步:使用Connection对象创建一个Session对象。
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 第五步:使用Session对象创建一个Destination对象。和发送端保持一致topic,并且话题的名称一致。
            Topic topic = session.createTopic("test-topic");
            // 第六步:使用Session对象创建一个Consumer对象。
            MessageConsumer consumer = session.createConsumer(topic);
            // 第七步:接收消息。
            consumer.setMessageListener(new MessageListener() {
    
                @Override
                public void onMessage(Message message) {
                    try {
                        TextMessage textMessage = (TextMessage) message;
                        String text = null;
                        // 取消息的内容
                        text = textMessage.getText();
                        // 第八步:打印消息。
                        System.out.println(text);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            System.out.println("topic的消费端03。。。。。");
            // 等待键盘输入
            System.in.read();
            // 第九步:关闭资源
            consumer.close();
            session.close();
            connection.close();
        }

    总结

    问题:linux下ActiveMQ闪退,java.net.UnknownHostException:root

    解决:vim /etc/hosts

    配置Ip和主机名

     

  • 相关阅读:
    Flask 应用上下文
    request 的介绍使用属性
    Session 常见操作
    cookie 常用操作
    Python三大web框架简单介绍
    Flask 入门
    类与类之间的关系
    request、response 中文乱码问题与解决方式
    JAVA笔记
    MySQL导入大sql 文件大小限制问题的解决
  • 原文地址:https://www.cnblogs.com/huozhonghun/p/10071156.html
Copyright © 2011-2022 走看看