Spring-ActiveMQ的点对点和Topic

yanghuashuiyue 2013-09-29 

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.example.activemq</groupId>
  <artifactId>activemq-test</artifactId>
  <version>1.0-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>activemq-test</name>
  <url>http://maven.apache.org</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <org.springframework.version>3.1.1.RELEASE</org.springframework.version>
  </properties>
  
<repositories>
  <repository>
	<id>kxcomm-maven</id>
	<name>Maven kxcomm Repository</name>
	<url>http://122.13.0.56:8088/nexus/content/groups/public/</url>
	<releases>
		<enabled>true</enabled>
	</releases>
	<snapshots>
		<enabled>true</enabled>
	</snapshots>
</repository>   
</repositories>

  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>
    <dependency>
			<groupId>commons-collections</groupId>
			<artifactId>commons-collections</artifactId>
			<version>3.2</version>
		</dependency>
		<dependency>
			<groupId>commons-configuration</groupId>
			<artifactId>commons-configuration</artifactId>
			<version>1.6</version>
		</dependency>
		<dependency>
			<groupId>commons-io</groupId>
			<artifactId>commons-io</artifactId>
			<version>1.3.2</version>
		</dependency>
		<dependency>
			<groupId>commons-logging</groupId>
			<artifactId>commons-logging</artifactId>
			<version>1.1.1</version>
		</dependency>
		<dependency>
			<groupId>log4j</groupId>
			<artifactId>log4j</artifactId>
			<version>1.2.17</version>
		</dependency>
		<dependency>
			<groupId>commons-beanutils</groupId>
			<artifactId>commons-beanutils</artifactId>
			<version>1.8.3</version>
		</dependency>
		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-asm</artifactId>
			<version>${org.springframework.version}</version>
		</dependency>
		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-jms</artifactId>
			<version>${org.springframework.version}</version>
		</dependency>
		<dependency>
			<groupId>com.davidkarlsen.commonstransaction.spring</groupId>
			<artifactId>commons-transaction-spring</artifactId>
			<version>0.9</version>
		</dependency>
		<dependency>
			<groupId>org.apache.httpcomponents</groupId>
			<artifactId>httpclient</artifactId>
			<version>4.2-beta1</version>
		</dependency>
		<dependency>
			<groupId>fastutil</groupId>
			<artifactId>fastutil</artifactId>
			<version>5.0.9</version>
		</dependency>

		<dependency>
			<groupId>ch.qos.logback</groupId>
			<artifactId>logback-classic</artifactId>
			<version>0.9.27</version>
			<scope>compile</scope>
		</dependency>
		<dependency>
			<groupId>ch.qos.logback</groupId>
			<artifactId>logback-core</artifactId>
			<version>0.9.27</version>
			<scope>compile</scope>
		</dependency>
		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-api</artifactId>
			<version>1.6.1</version>
			<scope>compile</scope>
		</dependency>
		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>jcl-over-slf4j</artifactId>
			<version>1.6.1</version>
			<scope>runtime</scope>
		</dependency>
		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-log4j12</artifactId>
			<version>1.7.5</version>
		</dependency>
		<dependency>
			<groupId>org.apache.activemq</groupId>
			<artifactId>activemq-all</artifactId>
			<version>5.8.0</version>
		</dependency>
  </dependencies>
</project>
<?xml version="1.0" encoding="UTF-8"?>
<beans
    xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:aop="http://www.springframework.org/schema/aop" 
    xmlns:tx="http://www.springframework.org/schema/tx" 
    xmlns:context="http://www.springframework.org/schema/context" 
    xsi:schemaLocation=" 
                http://www.springframework.org/schema/beans
                http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
                http://www.springframework.org/schema/aop
                http://www.springframework.org/schema/aop/spring-aop-2.5.xsd
                http://www.springframework.org/schema/tx
                http://www.springframework.org/schema/tx/spring-tx-2.5.xsd
                http://www.springframework.org/schema/context
                http://www.springframework.org/schema/context/spring-context-2.5.xsd"
                 default-autowire="byName" default-lazy-init="true">

<import resource="activemq-test.xml"/>

</beans>
<?xml version="1.0" encoding="UTF-8"?>

<beans
  xmlns="http://www.springframework.org/schema/beans"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
  http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
  http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

	<!-- 创建工厂连接 -->
	<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
		<property name="brokerURL" value="tcp://localhost:61616" />
	</bean>
	
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory" ref="connectionFactory" />
		 <property name="defaultDestination" ref="rantzDestination" />  
	</bean>
	
	<!-- Point-to-Point -->
	<!-- activeMQ消息目标 队列 -->
	<bean id="rantzDestination" class="org.apache.activemq.command.ActiveMQQueue">
		<constructor-arg index="0" value="rantz.marketing.queue"></constructor-arg>
	</bean>
	
	<!-- activeMQ消息目标 主题-->
        <!--	<bean id="rantzDestination" class="org.apache.activemq.command.ActiveMQTopic">-->
        <!--		<constructor-arg index="0" value="rantz.marketing.queue"></constructor-arg>-->
        <!--	</bean>-->
        
	<bean id="producer" class="activemq.test.p2p.producer.RantzMarketingGatewayImpl">
		<property name="jmsTemplate" ref="jmsTemplate" />
		<property name="destination" ref="rantzDestination" />
	</bean>
	<bean id="consumer" class="activemq.test.p2p.consumer.MarketingReceiverGatewayImpl">  
	    <property name="jmsTemplate" ref="jmsTemplate" />  
	</bean>  
	<!-- Point-to-Point End-->
	
	
	<!-- Topic -->
	<bean id="topic" class="org.apache.activemq.command.ActiveMQTopic" autowire="constructor">
        <constructor-arg index="0" value="kxcomm.mms.topic" />
    </bean>
    <bean id="control" class="org.apache.activemq.command.ActiveMQTopic" autowire="constructor">
        <constructor-arg index="0" value="kxcomm.mms.control" />
    </bean>
	
	<bean id="myListener" class="activemq.test.topic.MyListener">
    <property name="connectionFactory" ref="connectionFactory" />
    <property name="topic" ref="topic" />
    <property name="control" ref="control" />
    </bean>
   
   <bean id="myPublisher" class="activemq.test.topic.MyPublisher">
    <property name="connectionFactory" ref="connectionFactory" />
    <property name="topic" ref="topic" />
    <property name="control" ref="control" />
    </bean>
	<!-- Topic End-->

</beans>
package activemq.test.model;

import java.io.Serializable;

public class User implements Serializable{
	private static final long serialVersionUID = -3098636047897519268L;
	private String name;
	private String sex;
	private int age;
	public String getName() {
		return name;
	}
	public void setName(String name) {
		this.name = name;
	}
	public String getSex() {
		return sex;
	}
	public void setSex(String sex) {
		this.sex = sex;
	}
	public int getAge() {
		return age;
	}
	public void setAge(int age) {
		this.age = age;
	}
	@Override
	public String toString() {
		return "User [name=" + name + ", sex=" + sex + ", age=" + age + "]";
	}
	
}

PTP模型

PTP(Point-to-Point)模型是基于队列的，生产者发消息到队列，消费者从队列接收消息，队列的存在使得消息的异步传输成为可能。和邮件系统中的邮箱一样，队列可以包含各种消息，JMSProvider提供工具管理队列的创建、删除。JMSPTP模型定义了客户端如何向队列发送消息，从队列接收消息，浏览队列中的消息。

package activemq.test.p2p.consumer;

import org.springframework.jms.core.JmsTemplate;

import activemq.test.model.User;
public class MarketingReceiverGatewayImpl {
	
	private JmsTemplate jmsTemplate;
	
	public JmsTemplate getJmsTemplate() {
		return jmsTemplate;
	}
	public void setJmsTemplate(JmsTemplate jmsTemplate) {
		this.jmsTemplate = jmsTemplate;
	}
	
	public MarketingReceiverGatewayImpl() {
	}
	
	public void receiveMotorist() throws Exception{
		User message  = (User)jmsTemplate.receiveAndConvert();
		System.out.println("reviced msg is:" + message.toString());
	}
	
}
package activemq.test.p2p.consumer;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class StartConsumer {
	public static void main(String[] args) {
		/*开始加载spring配置文件*/
			ApplicationContext context = new ClassPathXmlApplicationContext("classpath:modules/applicationContext.xml");
			MarketingReceiverGatewayImpl rantzMarketingGateway= (MarketingReceiverGatewayImpl) context.getBean("consumer");
			System.out.println("Receive Start ...");
			try {
				while(true){
					rantzMarketingGateway.receiveMotorist();
				}
				
			} catch (Exception e) {
				e.printStackTrace();
			}
	    }
}
package activemq.test.p2p.producer;

public interface IRantzMarketingGateway {
	/**
	 * 
	 * 发送文本对象
	 * 
	 * @author zhangjh 新增日期：2013-9-20
	 * @since smsc-gateway
	 */
	public void sendMotoristInfo();

	/**
	 * 
	 * 发送对象
	 * 
	 * @author zhangjh 新增日期：2013-9-20
	 * @since smsc-gateway
	 */
	public void sendObjectInfo();
}
package activemq.test.p2p.producer;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

import activemq.test.model.User;

public class RantzMarketingGatewayImpl implements IRantzMarketingGateway {

	private JmsTemplate jmsTemplate;
	private Destination destination;

	public JmsTemplate getJmsTemplate() {
		return jmsTemplate;
	}

	public void setJmsTemplate(JmsTemplate jmsTemplate) {
		this.jmsTemplate = jmsTemplate;
	}

	public Destination getDestination() {
		return destination;
	}

	public void setDestination(Destination destination) {
		this.destination = destination;
	}

	public void sendMotoristInfo() {
		MessageCreator msg = new MessageCreator(){
			public Message createMessage(Session session) throws JMSException {
				return session.createTextMessage("这是一个测试,"+System.currentTimeMillis());
			}
		};
		jmsTemplate.send(destination, msg);
	}

	
	public void sendObjectInfo() {
		User u = new User();
		u.setAge(17);
		u.setName("yuky"+System.currentTimeMillis());
		u.setSex("女");
		jmsTemplate.convertAndSend(u);
		
	}
}
package activemq.test.p2p.producer;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class StartProducer {
	public static void main(String[] args) {
		/*开始加载spring配置文件*/
			ApplicationContext context = new ClassPathXmlApplicationContext("classpath:modules/applicationContext.xml");
			IRantzMarketingGateway rantzMarketingGateway= (RantzMarketingGatewayImpl) context.getBean("producer");
			for(int i=0;i<10;i++){
				rantzMarketingGateway.sendObjectInfo();
				System.out.println("Start ...");
			}
			
		}
}

PUB/SUB模型

消息订阅分为非持久订阅(non-durablesubscription)和持久订阅(durablesubscrip-tion)，非持久订阅只有当客户端处于激活状态，也就是和JMSProvider保持连接状态才能收到发送到某个主题的消息，而当客户端处于离线状态，这个时间段发到主题的消息将会丢失，永远不会收到。持久订阅时，客户端向JMS注册一个识别自己身份的ID，当这个客户端处于离线时，JMSProvider会为这个ID保存所有发送到主题的消息，当客户再次连接到JMSProvider时，会根据自己的ID得到所有当自己处于离线时发送到主题的消息。

package activemq.test.topic;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;

import activemq.test.model.User;

public class MyListener implements MessageListener {
	private ActiveMQConnectionFactory connectionFactory;
	private Connection connection;
	private Session session;
	private MessageProducer producer;
	private Topic topic;
    private Topic control;

	public Topic getTopic() {
		return topic;
	}

	public void setTopic(Topic topic) {
		this.topic = topic;
	}

	public Topic getControl() {
		return control;
	}

	public void setControl(Topic control) {
		this.control = control;
	}

	public ActiveMQConnectionFactory getConnectionFactory() {
		return connectionFactory;
	}

	public void setConnectionFactory(ActiveMQConnectionFactory connectionFactory) {
		this.connectionFactory = connectionFactory;
	}

	public void onMessage(Message message) {
		try{
			if (checkText(message, "SHUTDOWN")) {
	            try {
	                connection.close();
	                System.out.println("退出监听消息");
	            } catch (Exception e) {
	                e.printStackTrace(System.out);
	            }

	        } else if (checkText(message, "REPORT")) {
	            // send a report:
	            try {
	            	 System.out.println("MyListener->收到 a report");
	                long time = System.currentTimeMillis();
	                String msg = "MyListener->返回 a report :" + time + "ms";
	                System.out.println(msg);
	                producer.send(session.createTextMessage(msg));
	            } catch (Exception e) {
	                e.printStackTrace(System.out);
	            }
	        } else {
	        	ObjectMessage obj = (ObjectMessage)message;
	        	User u = (User) obj.getObject();
	        	System.out.println("Received  messages."+ u.toString());
	        }
		}catch(Exception e){
			
		}
	}

	public void run() throws JMSException {
		if(connectionFactory!=null){
			System.out.println("connectionFactory is ok");
			connection = connectionFactory.createConnection();
	        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	        MessageConsumer consumer = session.createConsumer(topic);
	        consumer.setMessageListener(this);
	        connection.start();
	        
	        producer = session.createProducer(control);
	        System.out.println("Waiting for messages...");
		}
	}
	
	private static boolean checkText(Message m, String s) {
        try {
            return m instanceof TextMessage && ((TextMessage)m).getText().equals(s);
        } catch (JMSException e) {
            e.printStackTrace(System.out);
            return false;
        }
    }
	
}
package activemq.test.topic;



import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class StartListener {

	public static void main(String[] args) {
		/*开始加载spring配置文件*/
			ApplicationContext context = new ClassPathXmlApplicationContext("classpath:modules/applicationContext.xml");
			MyListener myListener= (MyListener) context.getBean("myListener");
		
			try {
				if(myListener!=null){
					System.out.println("success...");
				}
				myListener.run();
				
			} catch (Exception e) {
				e.printStackTrace();
			}
	    }

}
package activemq.test.topic;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;

import activemq.test.model.User;

public class MyPublisher implements MessageListener {
	private ActiveMQConnectionFactory connectionFactory;
	private Connection connection;
	private Session session;
	private MessageProducer publisher;
	private Topic topic;
	private Topic control;
	private final Object mutex = new Object();

	public ActiveMQConnectionFactory getConnectionFactory() {
		return connectionFactory;
	}

	public void setConnectionFactory(ActiveMQConnectionFactory connectionFactory) {
		this.connectionFactory = connectionFactory;
	}

	public Topic getTopic() {
		return topic;
	}

	public void setTopic(Topic topic) {
		this.topic = topic;
	}

	public Topic getControl() {
		return control;
	}

	public void setControl(Topic control) {
		this.control = control;
	}

	
	public void onMessage(Message message) {
        synchronized (mutex) {
            System.out.println("Received report " + getReport(message) );
            
        }
    }
	
	Object getReport(Message m) {
        try {
            return ((TextMessage)m).getText();
        } catch (JMSException e) {
            e.printStackTrace(System.out);
            return e.toString();
        }
    }

	public void publish() throws Exception {
		 	User u = new User();
			u.setAge(17);
			u.setName("yuky"+System.currentTimeMillis());
			u.setSex("女");
	        // send events
			ObjectMessage obj = session.createObjectMessage();
			obj.setObject(u);
	        for (int i = 0; i < 10; i++) {
	            publisher.send(obj);
	            publisher.send(session.createTextMessage("REPORT"));
	        }
	    }
	 
	public void run() throws Exception {
		connection = connectionFactory.createConnection();
		session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		publisher = session.createProducer(topic);
		publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
		
		session.createConsumer(control).setMessageListener(this);
        connection.start();
	}
	
	public void stop() throws JMSException{
		publisher.send(session.createTextMessage("SHUTDOWN"));
		connection.stop();
        connection.close();
	}
}
package activemq.test.topic;

import javax.jms.JMSException;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class StartPublisher {
	public static void main(String[] args) throws InterruptedException {
		/*开始加载spring配置文件*/
			ApplicationContext context = new ClassPathXmlApplicationContext("classpath:modules/applicationContext.xml");
			MyPublisher publisher= (MyPublisher) context.getBean("myPublisher");
			try {
				publisher.run();
				publisher.publish();
			} catch (Exception e) {
				try {
					publisher.stop();
				} catch (JMSException e1) {
					e1.printStackTrace();
				}
				e.printStackTrace();
			}
			
    }
}
yanghuashuiyue

yanghuashuiyue

相关推荐

RabbitMQ都写了，RocketMQ怎么能落下？

整体架构最近看到了我在Github上写的rabbitmq-examples陆续被人star了，就想着写个rocketmq-examples。对rabbitmq感兴趣的小伙伴可以看我之前的文章。下面把RocketMQ的各个特性简单介绍一下，这样在用的时候心里

ljcsdn / 0评论 2020-07-27

Rabbitmq的使用四_Java Client方式使用Topic模式

topic交换机发送给队列的消息，通过routingkey进行匹配。　　* 号只能代替一个单词。　　# 号可以替换零个或多个单词。在这个例子中，我们将发送所有描述动物的信息。消息将与一个routingkey一起发送，routingkey由三个单词(两个点)

woaishanguosha / 0评论 2020-07-18

RocketMQ系列（三）消息的生产与消费

前面的章节，我们已经把RocketMQ的环境搭建起来了，是一个两主两从的异步集群。接下来，我们就看看怎么去使用RocketMQ，在使用之前，先要在NameServer中创建Topic，我们知道RocketMQ是基于Topic的消息队列，在生产者发送消息的时

qingyuerji / 0评论 2020-06-14

RocketMQ系列（四）顺序消费

折腾了好长时间才写这篇文章，顺序消费，看上去挺好理解的，就是消费的时候按照队列中的顺序一个一个消费；而并发消费，则是消费者同时从队列中取消息，同时消费，没有先后顺序。RocketMQ也有这两种方式的实现，但是在实践的过程中，就是不能顺序消费，好不容易能够实

MojitoBlogs / 0评论 2020-06-14

ActiveMQ

--消息队列连接池-->

xinglun / 0评论 2020-06-14

RocketMQ系列（三）消息的生产与消费

前面的章节，我们已经把RocketMQ的环境搭建起来了，是一个两主两从的异步集群。接下来，我们就看看怎么去使用RocketMQ，在使用之前，先要在NameServer中创建Topic，我们知道RocketMQ是基于Topic的消息队列，在生产者发送消息的时

MojitoBlogs / 0评论 2020-06-09

Shell 直接向kafka 的Topic发送消息

echo "hello" | kafka-console-producer.sh --broker-list node1:9092 --sync --topic sky. cat 文件 |kafka-console-producer.s

猫咪的一生 / 0评论 2020-06-03

kafka消息队列

Kafka适合离线和在线的消息消费,其消息保存在磁盘上。Kafka以Topic为单位进行消息的归纳,Producers向Topic发送消息,Consumers会消费预订了Topic的消息。多个Broker可以做成一个Cluster(集群)对外提供服务,每个

guicaizhou / 0评论 2020-05-05

Kafka核心技术与实战——08 | 最最最重要的集群参数配置（下）

答案就是 Topic 级别参数会覆盖全局 Broker 参数的值，而每个 Topic 都能设置自己的参数值，这就是所谓的 Topic 级别参数。retention.ms：规定了该 Topic 消息被保存的时长。retention.bytes：规定了要为该

猫咪的一生 / 0评论 2020-05-03

kafka基本命令和实践

./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 -partitions 1 --topic test

jiangkai00 / 0评论 2020-04-15

activeMQ中queue 与 topic 区别

3.queue模式生产者与消费者之间没有时间相关性，topic模式下生产者和消费者之间有一定的时间相关性，消费者只能接收到订阅之后的生产者发送的消息。

方新德 / 0评论 2020-04-08

kafka基础命令

　　消息生产者，发布消息到 kafka 集群的终端或服务。　　每条发布到 kafka 集群的消息属于的类别，即 kafka 是面向 topic 的。　　partition 是物理上的概念，每个 topic 包含一个或多个 partition。kafka 分

jiangkai00 / 0评论 2020-03-20

kafka术语

kafka broker：运行kafka server的服务被称为broker,它主要用于接手topic消息，管理topic,持久化和存储topic数据，kafka审计，kafka权限管理等。；kafka仅支持topic模式。topic partition

那年夏天0 / 0评论 2020-02-22

Kafka Topic Partition GroupId 及高可用

Topic主题用来区分不同类型的消息，实际也就是适用于不同的业务场景，默认消息保存一周时间；同一个Topic主题下，默认是一个partition分区，也就是只能有一个消费者来消费，如果想提升消费能力，就需要增加分区；消费者消费partition分区内的消息

sweetgirl0 / 0评论 2020-02-21

Kafka常用topic操作命令汇总

bin/kafka-topics.sh --create --zookeeper localhost:2181 --topic test_topic --replication-factor 3 --partitions 10 --config clean

sweetgirl0 / 0评论 2020-02-09

【kafka】常用命令行

topic的分区会被创建在当前主机，每个分区的副本会分配到集群的其他机器；# --zookeeper master:2181；信息写入zookeeper. $ kafka-topics.sh --create --zookeeper master:2181

MrZhangAdd / 0评论 2020-01-13

kafka的相关操作脚本

7、显示出Consumer的Group、Topic、分区ID、分区对应已经消费的Offset、logSize大小，Lag以及Owner等信息。最后再注意kafka的启动和停止。

jiangkai00 / 0评论 2020-01-12

golang nsq 同一个topic有多个channel,同时项目又互相引用事出现的问题

同一个topic下 不同channel分多个项目的时候,如果一个项目引用了别的项目包,那么被引用的项目就需要手动初始化mq,不能使用Init,因为会导致mq初始化两次,在一个channel下链接了两个消费者导致mq进行了负载均衡.

GoatSucker / 0评论 2020-01-11

认识kafka

kafka是一个高吞吐量,低延迟分布式的消息队列系统.kafka每秒可以处理几十万条消息,它的延迟最低只有几毫秒。kafka将所有消息组织成多个topic的形式存储,而每个topic又可以拆分成多个partition,每个partition又由一个一个消息

yangyutong00 / 0评论 2019-12-30

kafka（一）

//创建一个topic,名称为test_kafka，partitions为2个，replication有3个，topic的name为test2:. kafka-topics --zookeeper zk1:2181,zk2:2181,zk3:2181 --

amwayy / 0评论 2019-12-22

Kafka的安装与使用(转)

点对点消息系统：生产者发送一条消息到queue，一个queue可以有很多消费者，但是一个消息只能被一个消费者接受，当没有消费者可用时，这个消息会被保存直到有 一个可用的消费者，所以Queue实现了一个可靠的负载均衡。发布订阅消息系统：发布者发送到topic

amwayy / 0评论 2019-12-21

Rabbitmq的延时队列的使用

/* public static final String QUEUE = "queue";

lishijian / 0评论 2019-12-20

kafka没配置好，导致服务器重启之后，topic丢失，topic里面的消息也丢失

作为新手，我在本地配置了一个单机kafka，用得是kafka自带的zookeeper服务。tar -zxf kafka_2.10-0.10.1.0.tgz3）启动zookeeper，kafka. 4)新建一个TOPIC> bin/kafka-topi

guicaizhou / 0评论 2019-12-16

RabbitMQ官方教程五 Topic(GOLANG语言实现)

在上一教程中，我们改进了日志记录系统。我们没有使用只能进行虚拟广播的fanout交换器，而是使用直接交换器，并有可能选择性地接收日志。您可能从syslog unix工具中了解了这个概念，该工具根据严重性和工具路由日志。这将为我们提供很大的灵活性-我们可能只

cj0 / 0评论 2019-12-15

ActiveMQ、RabbitMQ、RocketMQ、Kafka有什么优点和缺点

　　　　现在社区以及国内应用都越来越少，官方社区现在对ActiveMQ 5.x维护越来越少，几个月才发布一个版本。　　　　erlang语言开发，性能极其好，延时很低；　　如果是大数据领域的实时计算、日志采集等场景，用Kafka是业内标准的，绝对没问题，社区

方新德 / 0评论 2019-12-14

关于Flink slot 和kafka topic 分区关系的说明

今天又有小伙伴在群里问 slot 和 kafka topic 分区的关系，大概回答了一下，这里整理一份。首先必须明确的是，Flink Task Manager 的 slot 数 和 topic 的分区数是没有直接关系的，而这个问题其实是问的是：任务的并发数

yuchuanchen / 0评论 2019-12-11

kafka 相关命令 偏移重置

kafka-topics.sh --create --zookeeper 10.202.13.196:2181 --replication-factor 1 --partitions 1 --topic cjw-test. kafka-producer-

猫咪的一生 / 0评论 2019-12-10

URLSearchParams对象

URLSearchParams对象用于处理URL中查询字符串，即？其实例对象的用法和Set数据结构类似。实例对象本身是可遍历对象。但是不是遍历器。下面的方法调用都返回一个Iterator对象。想要获取预想的结果，需要使用扩展运算符。而且和entries()

bapinggaitianli / 0评论 2019-12-06

1.kafka基础架构

Kafka是一个分布式的基于发布/订阅模式的消息队列，主要应用于大数据实时处理领域。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。一说到分布式就应该想到多台机器，并且更应该想到的就是分区，kafk

那年夏天0 / 0评论 2019-12-05

Kafka SQL查询Topic中消息内容

Kafka Eagle监控系统是一款用来监控Kafka集群的工具，目前更新的版本是v1.4.0，支持管理多个Kafka集群、管理Kafka主题、消费者组、消费者实例监控、消息阻塞告警、Kafka集群健康状态查看等等。关于Kafka Eagle 可视化监控工

那年夏天0 / 0评论 2019-11-21

Kafka的内部topic：__consumer_offsets 消费位移量主题

默认情况下__consumer_offsets有50个分区。下面是在windows平台上进行演示的，Linux 平台的命令基本一致，只有命令行路径以及后缀有差异。这两个参数的默认值分别是50和1，表示该topic有50个分区，副本因子是1。其中，curre

amwayy / 0评论 2019-11-20

kafka 入门详解

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。该项目的目标是为处理实时数据提供一个统一、高吞吐、低延迟的平台。其持久化层本质上是一个“按照分布式事务日志架构的大规模发布/订阅消息队列”，这使它作为企业级基础设施来

xpg / 0评论 2019-11-19

kafka topic命令

--config <name=value> A topic configuration override for the. --delete-config <name> A topic config

amwayy / 0评论 2015-07-02

kafka replication tools

如果这个preferred replica不是leader且在isr中，则controller发出request到这个preferred replica的broker，通知其作为topic partition的leader. The controller

lybingo / 0评论 2015-07-01

Kafka初识

　　　　1.1、Apache Kafka是一个 分布式 流处理 平台；　　　　2.1、构造 实时 流数据管道，在 系统 或 应用之间 可靠地获取数据。　　　　2.2、构建 实时 流式应用程序，对这些流数据进行转换或者影响。　　　　　　　　3.1.1、Kaf

guicaizhou / 0评论 2019-11-12

【CDN+】Kafka 的初步认识与入门

项目中用到了Kafka 这种分布式消息队列来处理日志，本文将对Kafka的基本概念和原理做一些简要阐释。消息处理方式有点对点，发布-订阅模式，Kafka就是一种发布-订阅模式。支持Kafka Server间的消息分区，及分布式消费，同时保证每个partit

猫咪的一生 / 0评论 2019-11-08

Kafka 的一些知识点整理【1】

Kafka集群包含一个或多个服务器，这种服务器被称为broker。broker端不维护数据的消费状态，提升了性能。副本，Kafka中同一条消息拷贝到多个地方做数据冗余，这些地方就是副本，副本分为Leader和Follower，角色不同作用不同，副本是对Pa

guicaizhou / 0评论 2019-11-06

Flume + kafka + HDFS构建日志采集系统

Flume是一个非常优秀日志采集组件，类似于logstash，我们通常将Flume作为agent部署在application server上，用于收集本地的日志文件，并将日志转存到HDFS、kafka等数据平台中；关于Flume的原理和特性，我们稍后详解，

amwayy / 0评论 2016-11-25

redis的发布订阅功能

redis提供了简单的发布订阅功能，对于一些合适的场景，比起专业的MQ来说，用起来更简单些。本文主要是记录下怎么在SpringBoot里头使用redis的发布订阅功能。private final AtomicLong counter = new Atomi

ljz0 / 0评论 2019-10-17

Activemq配置Jaas认证

指定了实用JAAS插件管理权限, "activemq" 一定要与login.config文件中的配置对应。四.groups.properties，内容格式groupname=username1,username2,username3，根

Intlgj / 0评论 2017-09-29

Kafka介绍

特点：生产者消费者模式，先进先出保证顺序，自己不丢数据，默认每隔7天清理数据，高吞吐量，没有主从关系，依靠Zk协调。　　producer只需要关心消息发往哪个topic，而consumer只关心自己订阅哪个topic，并不关心每条消息存于整个集群的哪个br

猫咪的一生 / 0评论 2019-11-01

kafka

./kafka-producer-perf-test --messages 500000 --message-size 1000 --batch-size 500 --topics test --threads 2 --broker-list localh

猫咪的一生 / 0评论 2019-10-30

Kafka管理与监控——彻底删除topic

如果需要被删除topic 此时正在被程序 produce和consume，则这些生产和消费程序需要停止。调用kafka delete命令则无法删除该topic。如果设置为true，则produce或者fetch 不存在的topic也会自动创建这个topic

yangyutong00 / 0评论 2019-10-27

kafka深度研究之路(5)-kafka新版常用命令汇总

6/启动终端的消费者 //总结：–from-beginning参数如果有表示从最开始消费数据，旧的和新的数据都会被消费，而没有该参数表示只会消费新产生的数据kafka-console-consumer.sh --bootstrap-server k

猫咪的一生 / 0评论 2019-10-24

Kafka之啥也没写

1 Kafka概念一个分布式消息队列，具有高性能、持久化、多副本备份、横向扩展能力。为了做到水平扩展，一个topic实际是由多个partition组成的，遇到瓶颈时，可以通过增加partition的数量来进行横向扩容。单个parition内是保证消息

Kafka / 0评论 2019-09-08

kafka+zookeeper集群搭建

注意，上面的博客不是高版本kafka，高版本的kafka不在支持zookeeper参数了，如下面这个例子

iamdll / 0评论 2019-05-27

KAFKA重新安装注意的问题

删除kafka相关文件rm-rf/kafka-logs/ --kafka topic存储路径rm-rf/etc/kafkarm-rf/var/log/kafkarm-rf/var/run/kafka/. 你可以通过命令：./bin/kafka-topic

王谦 / 0评论 2018-06-06

KAFKA

Kafka核心组件　　Topic：消息根据Topic进行归类，可以理解为一个队里。　　Producer：消息生产者，就是向kafka broker发消息的客户端。　　Consumer：消息消费者，向kafka broker取消息的客户端。　　broker：

jiaomrswang / 0评论 2018-06-05

kafka+zookeeper集群搭建

注意，上面的博客不是高版本kafka，高版本的kafka不在支持zookeeper参数了，如下面这个例子

ZooKeeper / 0评论 2019-05-27

从 Apache RocketMQ 和 Kafka 看 Topic 数量对单机性能的影响

对比发送端、接收端共存情况下，Topic数量对Kafka、RocketMQ的性能影响，分区数采用8个分区。在消息发送端，消费端共存的场景下，随着Topic数的增加Kafka吞吐量会急剧下降，而RocketMQ则表现稳定。

kobewwf / 0评论 2017-09-08